diff --git a/DESIGN.rst b/DESIGN.rst
index 9d22d20a..53b35087 100644
--- a/DESIGN.rst
+++ b/DESIGN.rst
@@ -1,30 +1,16 @@
 Project Status
 ==============
 
-Retired by author, looking for new owner :) or some help!
+This project is basically complete. At this time we're mostly working on the "telix" dependency,
+which provides a full application to BBS and MUDs and their protocols and features, which sometimes
+discovers fixes or feature gaps in telnetlib3.
+
+Below are various notes, sometimes trimmed from code, or README.rst, some of it is important but
+some of it just needs review and deletion, it is not all very accurate.
 
 Design
 ======
 
-reduce
-------
-
-outer telnetlib3-server and telnetlib3-client and examples should connect
-as exit(main(\*\*parse_args(sys.argv))), the _transform_args() function is
-rather shoe-horned, main() should declare keywords.
-
-**this is completed for server, copy to client**
-
-
-BaseTelnetProtocol
-------------------
-
-base_client.py and base_server.py actually share the same ABC
-base_protocol.py, they are almost mirror images of one another,
-which is pretty great, actually, so they can be reduced to
-BaseTelnetProtocol.
-
-
 On Linemode
 -----------
 
@@ -43,15 +29,22 @@ interface designed.
     comprehensive are our tests, and how well is our SLC working?
   - IAC-SB-LINEMODE-DO-FORWARDMASK is unhandled, raises NotImplementedError
 
+This project is the only known Server-side implementation of *Special Linemode
+Character* (SLC) negotiation and *Remote line editing* (`rfc-1184`_), other than
+BSD telnet, which was used as a guide for the bulk of this python implementation.
+
+Remote line editing is a comprehensive approach to providing responsive,
+low-latency output of characters received over slow network links, allowing
+incomplete lines to be buffered, while still providing remote editing
+facilities, such as backspace, kill line, etc.
+
+The Server and Client agree on a series of Special Linemode Character (SLC)
+function values, to agree on the keyboard characters used for Backspace,
+Interrupt Process (``^C``), Repaint (``^R``), Erase Word (``^W``), etc.
+
 TelnetWriter and TelnetServer
 -----------------------------
 
-feed_byte called by telnet server should be a coroutine
-receiving data by send. It should yield out-of-bound values, None otherwise?
-'is_oob', or 'slc_received', etc.?  We're still considering ... the state still
-requires tracking, but this would turn multiple function calls into a .send()
-into generator, better for state loops or bandwidth, maybe?
-
 handle_xon resumes writing in a way that is not obvious -- we should
 be using the true 'pause_writing' and 'resume_writing' methods of our
 base protocol.  The given code was written before these methods became
@@ -60,6 +53,7 @@ availabilities.
 
 On STATUS rfc
 -------------
+
 We've seen everything negotiate fine, but what exactly are we expected to do
 when the distant end's concept of our negotiation STATUS disagrees with our
 own? Match theirs, should we re-negotiate or re-affirm misunderstood values?
@@ -81,97 +75,14 @@ SLC flush
 - SLC flushin/flushout attributes are not honored.  Not entirely sure
   how to handle these two values with asyncio yet.
 
-
-
-telsh
-=====
-
-In addition to remote line editing as described below, a pure-python shell,
-*telsh* is provided to allow toggling of server options and session parameters.
-In this way, it provides a suitable interface for testing telnet client
-capabilities.
-
-It is only in the interest of this project to provide enough shell-like
-capabilities to demonstrate remote line editing and an extensible environment
-for session introspection. An example of this is assigning a new value to
-CHARSET, toggling in and outbinary, thereby enabling UTF8 input/output, etc.
-
 UTF8
 ====
 
-CHARSET (`rfc-2066`_) specifies a codepage, not an encoding. At the time, this
-was more or less limited to specifying the codepage used to display bytes of the
-range 127 through 255.  Unimplemented in BSD client, and generally found
-implemented only in recent MUD client (Atlantis_) and servers. Most common
-values are: ASCII, UTF8, BIG5, and LATIN1.
-
-The default preferred encoding for clients that negotiate BINARY but not
-CHARSET, such as the BSD client, is defined by the TelnetServer keyword
-argument ``default_encoding`` ('UTF8' by default).
-
-The example shell *telsh* allows changing encoding on the fly by setting the
-'CHARSET' session environment value at the *telsh* command prompt by issuing
-command::
-
-    set CHARSET=UTF8
-
-Setting binary for only a single direction ('outbinary' or 'inbinary') is
-supported. Client support of one does not immediately toggle the other, it
-must be negotiated both ways for full UTF8 input and output.
-
 Some clients (`TinTin++`_) incorrectly negotiation either directions (WILL,
 DO/WONT, DONT) as a single option, causing only one reply for a request of
 either 'outbinary' or 'inbinary' for which it always declines, only once, for
 either request (Even when configured for UTF8).
 
-CP437
-=====
-
-Additionally, a contrib.cp437 module is included (authored by tehmaze_) which
-translates output meant to be translated by DOS Emulating programs to their
-comparable UTF-8 font. This is used by argument *--cp437* of the telnet-client_
-program.
-
-Some bulletin-board systems will send extended ascii characters (such as those
-used by 
-
-Telnet
-======
-
-The Telnet protocol is over 40 years old and still in use today. Telnet predates
-TCP, and was used over a wide array of transports, especially on academic and
-military systems. Nearly all computer networking that interacted with human
-interfaces was done using the Telnet protocol prior to the mass-adoption of
-the World Wide Web in the mid 1990's, when SSH became more commonplace.
-
-Naturally, Telnet as a code project inevitably must handle a wide variety of
-connecting clients and hosts, due to limitations of their networking Transport
-, Terminals, their drivers, and host operating systems.
-
-This implementation aims to implement only those capabilities "found in the
-wild", and includes, or does not include, mechanisms that are suitable only
-for legacy or vendor-implemented options. It even makes one of its own: the
-encoding' used in binary mode is the value replied by the CHARSET negotiation
-(`rfc-2066`_).
-
-
-
-Remote LineMode
----------------
-
-This project is the only known Server-side implementation of *Special Linemode
-Character* (SLC) negotiation and *Remote line editing* (`rfc-1184`_), other than
-BSD telnet, which was used as a guide for the bulk of this python implementation.
-
-Remote line editing is a comprehensive approach to providing responsive,
-low-latency output of characters received over slow network links, allowing
-incomplete lines to be buffered, while still providing remote editing
-facilities, such as backspace, kill line, etc.
-
-The Server and Client agree on a series of Special Linemode Character (SLC)
-function values, to agree on the keyboard characters used for Backspace,
-Interrupt Process (``^C``), Repaint (``^R``), Erase Word (``^W``), etc.
-
 Kludge Mode
 -----------
 
@@ -458,8 +369,8 @@ TODO
 - xon/xoff is unimplemented, see
   telnetlib3.stream_writer.TelnetWriter.handle_xon and handle_xoff.
 
-- After long-running (~2mo) job of telnetlib3 server on public IP, we ran
-  out of memory ! write test verifying garbage collects!
+- SLC flushin/flushout attributes are not honored.  Not entirely sure
+  how to handle these two values with asyncio yet.
 
 - TelnetReader has no need for declaring server/client=True, it behaves the
   same either way.
@@ -474,17 +385,6 @@ TODO
   would return a line BEGINNING with either LF or NUL when the previous line
   ended with CR, we simply discard that byte.
  
-- base_client.py and base_server.py actually share the same ABC
-  base_protocol.py, they are almost mirror images of one another,
-  which is pretty great, actually.  just reduce.
-
-- ValueError is used for many places where, the error is indicating that
-  a negotiation state that was attempted by the remote end is invalid,
-  for example: "received IAC SB LFLOW without first receiving IAC DO LFLOW."
-
-- SLC flushin/flushout attributes are not honored.  Not entirely sure
-  how to handle these two values with asyncio yet.
-
 - LINEMODE compliance needs a lot of work.
   - possibly, we remove LINEMODE support entirely. I only know of one client,
     BSD telnet, that is capable of negotiating -- this is the C code from which
@@ -495,7 +395,6 @@ TODO
     comprehensive are our tests, and how well is our SLC working?
   - IAC-SB-LINEMODE-DO-FORWARDMASK is unhandled, raises NotImplementedError
 
-    
 - _receive_status(self, buf) response to STATUS does not *honor* given state
    values. only a non-compliant distant end would cause such a condition. so
    it is decided to leave it as "conflict report only, no action always"
diff --git a/bin/server_mud.py b/bin/server_mud.py
index 0400a34b..c7eae427 100755
--- a/bin/server_mud.py
+++ b/bin/server_mud.py
@@ -23,7 +23,7 @@
 from typing import Any
 
 # local
-from telnetlib3.telopt import GMCP, MSDP, MSSP, WILL
+from telnetlib3.telopt import ZMP, GMCP, MSDP, MSSP, WILL
 from telnetlib3.server_shell import readline2
 
 log = logging.getLogger("mud")
@@ -274,6 +274,11 @@ def on_gmcp(writer: Any, package: str, data: Any) -> None:
     writer.write(f"[DEBUG GMCP] {package}: {json.dumps(data)}\r\n")
 
 
+def on_zmp(command: str, *args: str) -> None:
+    """Handle incoming ZMP from a client."""
+    log.debug("ZMP: %s %r", command, args)
+
+
 def get_msdp_var(player: Player, var: str) -> dict[str, Any] | None:
     """Return MSDP value dict for *var*, or ``None`` if unknown."""
     if var == "CHARACTER_NAME":
@@ -757,8 +762,10 @@ async def shell(reader: Any, writer: Any) -> None:
     writer.iac(WILL, GMCP)
     writer.iac(WILL, MSDP)
     writer.iac(WILL, MSSP)
+    writer.iac(WILL, ZMP)
     writer.set_ext_callback(GMCP, lambda pkg, data: on_gmcp(writer, pkg, data))
     writer.set_ext_callback(MSDP, lambda variables: on_msdp(writer, variables))
+    writer.set_ext_callback(ZMP, on_zmp)
     ssl_obj = writer.get_extra_info("ssl_object")
     if ssl_obj is not None:
         version = ssl_obj.version() or "TLS"
diff --git a/docs/history.rst b/docs/history.rst
index 989140e4..be8835f3 100644
--- a/docs/history.rst
+++ b/docs/history.rst
@@ -1,5 +1,34 @@
 History
 =======
+
+5.0.0
+  * changed: :meth:`~telnetlib3.stream_writer.TelnetWriter.handle_zmp` now receives ``command,
+    *args`` instead of one ``parts`` list; ``zmp_data`` moved to ``writer.ctx`` and is now a dict
+    keyed by command (was a list of messages).  New
+    :meth:`~telnetlib3.stream_writer.TelnetWriter.send_zmp`.
+  * changed: MUD protocol subnegotiation data (``mssp_data``, ``atcp_data``, ``aardwolf_data``,
+    ``mxp_data``, ``comport_data``) moved from :class:`~telnetlib3.stream_writer.TelnetWriter` to
+    :class:`~telnetlib3._session_context.TelnetSessionContext` (``writer.ctx.mssp_data``, etc.).
+    Deprecated writer properties delegate to ``ctx``; ``writer.zmp_data`` is removed without
+    deprecation.
+  * changed: client-side MUD protocol declines (GMCP, MSDP, MSSP, MSP, MXP, ZMP, AARDWOLF, ATCP)
+    remain the default; the decline log messages now name the enable mechanism (``always_will`` /
+    ``always_do`` / ``passive_do``) used by downstream clients such as telix to accept them.  Note
+    that MXP's negotiation direction is server-sends-``DO`` (the LPMud family convention, e.g.
+    Discworld), so a client accepting MXP replies ``WILL`` to ``IAC DO MXP``.
+  * new: :meth:`~telnetlib3.stream_writer.TelnetWriter.add_will_callback` and
+    :meth:`~telnetlib3.stream_writer.TelnetWriter.remove_will_callback` for per-option callbacks
+    invoked after :meth:`~telnetlib3.stream_writer.TelnetWriter.handle_will` negotiation.  Replaces
+    the previous closure-wrapping pattern in :class:`~telnetlib3.client.TelnetClient` for GMCP, ZMP,
+    and CHARSET will-detection.
+  * enhancement: ``telnetlib3-fingerprint`` now accepts all MUD protocol offers (ATCP, AARDWOLF,
+    MSP, MXP, MSDP, MSSP) to collect subnegotiation data.
+  * enhancement: sub-negotiation payloads are bounded to 1,000KB.
+  * enhancement: ``--loglevel=trace`` receive dumps show the decompressed telnet stream (MCCP2,
+    MCCP3) instead of raw compressed bytes.
+  * bugfix: ``IAC SB IAC SE`` (sub-negotiation with no option byte) should not raise ``IndexError``
+
+
 4.0.6
   * bugfix: default GMCP modules requested are now in lowercase instead of titlecase
 
diff --git a/docs/rfcs.rst b/docs/rfcs.rst
index 016bf327..4e8a3dd1 100644
--- a/docs/rfcs.rst
+++ b/docs/rfcs.rst
@@ -77,6 +77,9 @@ Dungeon) servers and clients.
 * `GMCP`_ (Generic MUD Communication Protocol, option 201). JSON-based
   bidirectional messaging for game data such as room info, character vitals,
   and client metadata.
+* `ZMP`_ (Zenith Mud Protocol, option 93). Bidirectional messaging of
+  NUL-delimited string lists (a command plus arguments), carrying data such
+  as room info and character vitals.
 * `MSDP`_ (MUD Server Data Protocol, option 69). Structured key-value protocol
   for game variables with support for nested tables and arrays.
 * `MSSP`_ (MUD Server Status Protocol, option 70). Server metadata protocol
@@ -93,13 +96,14 @@ Dungeon) servers and clients.
 .. _MSSP: https://tintin.mudhalla.net/protocols/mssp/
 .. _MCCP2: https://tintin.mudhalla.net/protocols/mccp/
 .. _MCCP3: https://tintin.mudhalla.net/protocols/mccp/
+.. _ZMP: https://discworld.starturtle.net/external/protocols/zmp.html
 
 MUDs Not Implemented
 --------------------
 
 Constants are also defined for the following MUD options, though their handlers
-are not implemented: MCCP (85, legacy compression), MXP (91, markup), ZMP
-(93, messaging), MSP (90, sound), and ATCP (200, Achaea-specific).
+are not implemented: MCCP (85, legacy compression), MXP (91, markup), MSP (90,
+sound), and ATCP (200, Achaea-specific).
 
 Additional Resources
 --------------------
diff --git a/pyproject.toml b/pyproject.toml
index d1e4ccb2..5d4ed9ae 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "telnetlib3"
-version = "4.0.6"  # Keep in sync with telnetlib3/accessories.py::get_version !
+version = "5.0.0"  # Keep in sync with telnetlib3/accessories.py::get_version !
 description = " Python Telnet server and client CLI and Protocol library"
 readme = "README.rst"
 license = "ISC"
diff --git a/telnetlib3/_session_context.py b/telnetlib3/_session_context.py
index cacc0ffb..6dd34841 100644
--- a/telnetlib3/_session_context.py
+++ b/telnetlib3/_session_context.py
@@ -53,3 +53,10 @@ def __init__(
         self.autoreply_wait_fn = autoreply_wait_fn
         self.typescript_file = typescript_file
         self.gmcp_data: dict[str, Any] = gmcp_data if gmcp_data is not None else {}
+        self.zmp_data: dict[str, list[str]] = {}
+        # MUD protocol data moved from TelnetWriter to ctx for consistency
+        self.mssp_data: Optional[dict[str, str | list[str]]] = None
+        self.atcp_data: list[tuple[str, str]] = []
+        self.aardwolf_data: list[dict[str, Any]] = []
+        self.mxp_data: list[bytes] = []
+        self.comport_data: Optional[dict[str, Any]] = None
diff --git a/telnetlib3/accessories.py b/telnetlib3/accessories.py
index 078fb1b1..34ccb039 100644
--- a/telnetlib3/accessories.py
+++ b/telnetlib3/accessories.py
@@ -42,7 +42,7 @@
 
 def get_version() -> str:
     """Return the current version of telnetlib3."""
-    return "4.0.6"  # keep in sync with pyproject.toml !
+    return "5.0.0"  # keep in sync with pyproject.toml !
 
 
 def encoding_from_lang(lang: str) -> Optional[str]:
diff --git a/telnetlib3/client.py b/telnetlib3/client.py
index 9eb0c83d..08add246 100755
--- a/telnetlib3/client.py
+++ b/telnetlib3/client.py
@@ -12,7 +12,7 @@
 import asyncio
 import argparse
 import functools
-from typing import Any, Dict, List, Tuple, Union, Callable, Optional, Sequence
+from typing import Any, Dict, List, Tuple, Union, Callable, Optional, Sequence, Collection
 
 # local
 from telnetlib3 import accessories, client_base
@@ -73,6 +73,8 @@ def __init__(
         waiter_closed: Optional[asyncio.Future[None]] = None,
         _waiter_connected: Optional[asyncio.Future[None]] = None,
         gmcp_modules: Optional[List[str]] = None,
+        zmp_check_handler: Optional[Callable[[str], bool]] = None,
+        zmp_supported_commands: Optional[Collection[str]] = None,
     ) -> None:
         """Initialize TelnetClient with terminal parameters."""
         self._compression = compression
@@ -89,6 +91,11 @@ def __init__(
         )
         self._gmcp_modules = gmcp_modules or list(_DEFAULT_GMCP_MODULES)
         self._gmcp_hello_sent = False
+        self._zmp_check_handler = zmp_check_handler  # None means refuse all
+        self._zmp_ident_sent = False
+        self._zmp_supported_commands = (
+            set(zmp_supported_commands) if zmp_supported_commands else set()
+        )
         self._send_environ = set(send_environ or self.DEFAULT_SEND_ENVIRON)
         self._extra.update(
             {
@@ -144,62 +151,74 @@ def connection_made(self, transport: asyncio.BaseTransport) -> None:
         ):
             self.writer.set_ext_offer_callback(opt, offer_func)
 
-        # Override the default handle_will method to detect when both sides support CHARSET
-        # Store the original only on first connection to prevent chain growth on reconnect.
-        if not hasattr(self.writer, "_original_handle_will"):
-            self.writer._original_handle_will = self.writer.handle_will
-        else:
-            self.writer.handle_will = (  # type: ignore[method-assign]
-                self.writer._original_handle_will
-            )
-        original_handle_will = self.writer.handle_will
-        writer = self.writer
+        self.writer.add_will_callback(CHARSET, self.on_will_charset)
 
-        def enhanced_handle_will(opt: bytes) -> None:
-            original_handle_will(opt)
+        self.setup_gmcp()
+        self.setup_zmp()
 
-            # If this was a WILL CHARSET from the server, and we also have WILL CHARSET enabled,
-            # log that both sides support CHARSET. The server should initiate the actual REQUEST.
-            if (
-                opt == CHARSET
-                and writer.remote_option.enabled(CHARSET)
-                and writer.local_option.enabled(CHARSET)
-            ):
-                self.log.debug("Both sides support CHARSET, ready for server to initiate REQUEST")
+    def on_will_charset(self, opt: bytes) -> None:
+        """Log when both sides support CHARSET after WILL negotiation."""
+        from telnetlib3.telopt import CHARSET
 
-        self.writer.handle_will = enhanced_handle_will  # type: ignore[method-assign]
+        if self.writer.remote_option.enabled(CHARSET) and self.writer.local_option.enabled(CHARSET):
+            self.log.debug("Both sides support CHARSET, ready for server to initiate REQUEST")
 
-        self._setup_gmcp()
-
-    def _setup_gmcp(self) -> None:
+    def setup_gmcp(self) -> None:
         """Wire GMCP callback and WILL-detection for Core.Hello handshake."""
         from telnetlib3.telopt import GMCP
 
+        self.writer.passive_do.add(GMCP)
         self.writer.set_ext_callback(GMCP, self.on_gmcp)
+        self.writer.add_will_callback(GMCP, self.on_will_gmcp)
 
-        # Capture current handle_will (already includes CHARSET wrapper).
-        # On reconnect, _original_handle_will was already restored in connection_made,
-        # so this always wraps exactly once.
-        original_handle_will_gmcp = self.writer.handle_will
+    def on_will_gmcp(self, opt: bytes) -> None:
+        """Send Core.Hello after GMCP negotiation."""
+        from telnetlib3.telopt import GMCP
 
-        def _detect_gmcp_will(opt: bytes) -> None:
-            original_handle_will_gmcp(opt)
-            if opt == GMCP and self.writer.remote_option.enabled(GMCP):
-                self._send_gmcp_hello()
+        enabled = self.writer.remote_option.enabled(GMCP)
+        hello_sent = self._gmcp_hello_sent
+        self.log.debug("on_will_gmcp: remote_enabled=%s _gmcp_hello_sent=%s", enabled, hello_sent)
+        if enabled:
+            self.send_gmcp_hello()
 
-        self.writer.handle_will = _detect_gmcp_will  # type: ignore[method-assign]
+    def setup_zmp(self) -> None:
+        """Wire ZMP callback and WILL-detection for zmp.ident handshake."""
+        from telnetlib3.telopt import ZMP
 
-    def _send_gmcp_hello(self) -> None:
+        self.writer.passive_do.add(ZMP)
+        self.writer.set_ext_callback(ZMP, self.on_zmp)
+        self.writer.add_will_callback(ZMP, self.on_will_zmp)
+
+    def on_will_zmp(self, opt: bytes) -> None:
+        """Send zmp.ident after ZMP negotiation."""
+        from telnetlib3.telopt import ZMP
+
+        if self.writer.remote_option.enabled(ZMP):
+            self.send_zmp_ident()
+
+    def send_gmcp_hello(self) -> None:
         """Send ``Core.Hello`` and ``Core.Supports.Set`` after GMCP negotiation."""
         if self._gmcp_hello_sent:
             return
         self._gmcp_hello_sent = True
         from telnetlib3.accessories import get_version
 
+        modules = [m.lower() for m in self._gmcp_modules]
         self.writer.send_gmcp("Core.Hello", {"client": "telnetlib3", "version": get_version()})
-        wire_modules = [m.lower() for m in self._gmcp_modules]
-        self.writer.send_gmcp("Core.Supports.Set", wire_modules)
-        self.log.info("GMCP handshake: Core.Hello + Core.Supports.Set %s", wire_modules)
+        self.writer.send_gmcp("Core.Supports.Set", modules)
+        self.log.info("GMCP handshake: Core.Hello + Core.Supports.Set %s", modules)
+
+    def send_zmp_ident(self) -> None:
+        """Send ``zmp.ident`` after ZMP negotiation."""
+        if self._zmp_ident_sent:
+            return
+        self._zmp_ident_sent = True
+        from telnetlib3.accessories import get_version
+
+        self.writer.send_zmp("zmp.ident", "telnetlib3", get_version())
+        self.log.info("ZMP handshake: zmp.ident telnetlib3 %s", get_version())
+        for cmd in sorted(self._zmp_supported_commands):
+            self.writer.send_zmp("zmp.support", cmd)
 
     def on_gmcp(self, package: str, data: Any) -> None:
         """Store incoming GMCP data on ``writer.ctx``, merging dict updates."""
@@ -210,6 +229,39 @@ def on_gmcp(self, package: str, data: Any) -> None:
             gmcp[package] = data
         self.log.debug("GMCP: %s %r", package, data)
 
+    def on_zmp(self, command: str, *args: str) -> None:
+        """
+        Receive and dispatch a ZMP message.
+
+        Auto-responds to ``zmp.check`` and ``zmp.send-support``.
+        Stores latest value for each command in ``writer.ctx.zmp_data``.
+        """
+        self.log.debug("ZMP: %s %r", command, args)
+        self.writer.ctx.zmp_data[command] = list(args)
+        if command == "zmp.check" and args:
+            cmd = args[0]
+            self._respond_zmp_support(cmd)
+        elif command == "zmp.send-support":
+            if args:
+                for cmd in args:
+                    self._respond_zmp_support(cmd)
+            elif not self._zmp_ident_sent:
+                for cmd in sorted(self._zmp_supported_commands):
+                    self._respond_zmp_support(cmd)
+
+    def _respond_zmp_support(self, cmd: str) -> None:
+        """Send ``zmp.support`` or ``zmp.no-support`` for *cmd*."""
+        if self.zmp_check(cmd):
+            self.writer.send_zmp("zmp.support", cmd)
+        else:
+            self.writer.send_zmp("zmp.no-support", cmd)
+
+    def zmp_check(self, cmd: str) -> bool:
+        """Return True if the client supports the given ZMP command."""
+        if self._zmp_check_handler is not None:
+            return self._zmp_check_handler(cmd)
+        return False
+
     def send_ttype(self) -> str:
         """Callback for responding to TTYPE requests."""
         result: str = self._extra["term"]
@@ -689,7 +741,7 @@ def _patched_connection_made(transport: asyncio.BaseTransport) -> None:
                 client.writer.always_dont = always_dont
             from .telopt import GMCP as _GMCP
 
-            client.writer.passive_do = {_GMCP}
+            client.writer.passive_do.add(_GMCP)
             client.writer.environ_encoding = environ_encoding
             client.writer._encoding_explicit = encoding_explicit
 
diff --git a/telnetlib3/client_base.py b/telnetlib3/client_base.py
index c5e8d0e5..bd379146 100644
--- a/telnetlib3/client_base.py
+++ b/telnetlib3/client_base.py
@@ -226,8 +226,6 @@ def data_received(self, data: bytes) -> None:
         Buffer incoming data and schedule async processing to keep the event loop responsive. Apply
         read-side backpressure using transport.pause_reading()/resume_reading().
         """
-        if self.log.isEnabledFor(TRACE):
-            self.log.log(TRACE, "recv %d bytes\n%s", len(data), hexdump(data, prefix="<<  "))
         self._last_received = datetime.datetime.now()
 
         # Detect SyncTERM font switching sequences and auto-switch encoding.
@@ -354,6 +352,19 @@ def check_negotiation(self, final: bool = False) -> bool:
 
     # private methods
 
+    def _trace_recv(self, data: bytes, decompressed: bool = False) -> None:
+        """
+        Log received bytes at TRACE level.
+
+        :param data: Bytes received from the transport.
+        :param decompressed: True when *data* is the MCCP2-decompressed telnet stream; the raw
+            compressed bytes are not logged.
+        """
+        if not data or not self.log.isEnabledFor(TRACE):
+            return
+        label = " (decompressed)" if decompressed else ""
+        self.log.log(TRACE, "recv %d bytes%s\n%s", len(data), label, hexdump(data, prefix="<<  "))
+
     def _process_chunk(self, data: bytes) -> bool:
         """Process a chunk of received bytes; return True if any IAC/SB cmd observed."""
         self._last_received = datetime.datetime.now()
@@ -380,14 +391,15 @@ def _process_chunk(self, data: bytes) -> bool:
             if self._mccp2_decompressor.eof:
                 unused = self._mccp2_decompressor.unused_data
                 self._mccp2_end()
-                cmd = self._process_chunk_inner(data)
+                cmd = self._process_chunk_inner(data, decompressed=True)
                 if unused:
                     cmd = self._process_chunk(unused) or cmd
                 return cmd
+            return self._process_chunk_inner(data, decompressed=True)
 
         return self._process_chunk_inner(data)
 
-    def _process_chunk_inner(self, data: bytes) -> bool:
+    def _process_chunk_inner(self, data: bytes, decompressed: bool = False) -> bool:
         """Inner chunk processing with IAC interpretation and mid-chunk MCCP2 detection."""
         try:
             mode = self.writer.mode
@@ -408,9 +420,17 @@ def _process_chunk_inner(self, data: bytes) -> bool:
         if self.writer._compressed_remainder is not None:
             remainder = self.writer._compressed_remainder
             self.writer._compressed_remainder = None
+            if not decompressed:
+                # The chunk activating MCCP2 mixes plain telnet and compressed
+                # bytes; trace only the plain prefix here, the compressed
+                # remainder is traced after decompression by the recursive
+                # _process_chunk() call.
+                self._trace_recv(data[: len(data) - len(remainder)])
             self._mccp2_start()
             if remainder:
                 cmd_received = self._process_chunk(remainder) or cmd_received
+        else:
+            self._trace_recv(data, decompressed=decompressed)
 
         # MCCP3: start compressor when writer signals activation
         if self.writer.mccp3_active and self._mccp3_compressor is None:
diff --git a/telnetlib3/client_shell.py b/telnetlib3/client_shell.py
index 6fd8025b..418f97c6 100644
--- a/telnetlib3/client_shell.py
+++ b/telnetlib3/client_shell.py
@@ -579,7 +579,7 @@ async def _raw_event_loop(
                 if raw_mode is None and want_repl():
                     state.reactivate_repl = True
             stdout.write(out.encode())
-            if hasattr(stdout, 'drain'):
+            if hasattr(stdout, "drain"):
                 await stdout.drain()
             _ts_file = telnet_writer.ctx.typescript_file
             if _ts_file is not None:
diff --git a/telnetlib3/mud.py b/telnetlib3/mud.py
index 412ed378..299d09c2 100644
--- a/telnetlib3/mud.py
+++ b/telnetlib3/mud.py
@@ -40,6 +40,7 @@
     "mssp_encode",
     "mssp_decode",
     "MsdpParser",
+    "zmp_encode",
     "zmp_decode",
     "atcp_decode",
     "aardwolf_decode",
@@ -276,6 +277,24 @@ def mssp_decode(buf: bytes, encoding: str = "utf-8") -> dict[str, str | list[str
     return result
 
 
+def zmp_encode(command: str, *args: str) -> bytes:
+    r"""
+    Encode a ZMP message.
+
+    :param command: ZMP command name (e.g., ``"zmp.ident"``).
+    :param args: Zero or more argument strings.
+    :returns: NUL-delimited payload bytes suitable for ZMP subnegotiation.
+
+    Example::
+
+        >>> zmp_encode("zmp.ident", "telnetlib3", "1.0")
+        b'zmp.ident\\x00telnetlib3\\x001.0\\x00'
+    """
+    parts = [command.encode("utf-8")]
+    parts.extend(a.encode("utf-8") for a in args)
+    return b"\x00".join(parts) + b"\x00"
+
+
 def zmp_decode(buf: bytes, encoding: str = "utf-8") -> list[str]:
     """
     Decode ZMP payload to list of NUL-delimited strings.
diff --git a/telnetlib3/server_base.py b/telnetlib3/server_base.py
index ba59a98a..72c69d04 100644
--- a/telnetlib3/server_base.py
+++ b/telnetlib3/server_base.py
@@ -188,14 +188,25 @@ def begin_shell(self, future: asyncio.Future[None]) -> None:
                 loop = asyncio.get_event_loop()
                 loop.create_task(coro)
 
+    def _trace_recv(self, data: bytes, decompressed: bool = False) -> None:
+        """
+        Log received bytes at TRACE level.
+
+        :param data: Bytes received from the transport.
+        :param decompressed: True when *data* is the MCCP3-decompressed telnet stream; the raw
+            compressed bytes are not logged.
+        """
+        if not data or not logger.isEnabledFor(TRACE):
+            return
+        label = " (decompressed)" if decompressed else ""
+        logger.log(TRACE, "recv %d bytes%s\n%s", len(data), label, hexdump(data, prefix="<<  "))
+
     def data_received(self, data: bytes) -> None:
         """
         Process bytes received by transport.
 
         Feeds raw bytes through the writer's IAC interpreter, forwarding in-band data to the reader.
         """
-        if logger.isEnabledFor(TRACE):
-            logger.log(TRACE, "recv %d bytes\n%s", len(data), hexdump(data, prefix="<<  "))
         self._last_received = datetime.datetime.now()
         self._rx_bytes += len(data)
 
@@ -207,12 +218,15 @@ def data_received(self, data: bytes) -> None:
                 logger.warning("MCCP3 decompression error, disabling")
                 self._mccp3_end()
                 return
+            self._trace_recv(data, decompressed=True)
             if self._mccp3_decompressor.eof:
                 unused = self._mccp3_decompressor.unused_data
                 self._mccp3_end()
                 if unused:
                     self.data_received(unused)
                     return
+        else:
+            self._trace_recv(data)
 
         if self.writer.slc_simulated:
             slc_vals = {defn.val[0] for defn in self.writer.slctab.values() if defn.val != theNULL}
diff --git a/telnetlib3/server_fingerprinting.py b/telnetlib3/server_fingerprinting.py
index da401252..01c87081 100644
--- a/telnetlib3/server_fingerprinting.py
+++ b/telnetlib3/server_fingerprinting.py
@@ -513,6 +513,13 @@ async def _fingerprint_session(
     start_time = time.time()
     cursor = _VirtualCursor(encoding=writer.environ_encoding)
 
+    # Accept all MUD protocol WILL requests so we can collect
+    # subnegotiation data (ATCP, AARDWOLF, MXP, MSP, etc.).
+    from telnetlib3.telopt import MSP, MXP, ATCP, MSDP, MSSP, AARDWOLF
+
+    for opt in (ATCP, AARDWOLF, MSP, MXP, MSDP, MSSP):
+        writer.passive_do.add(opt)
+
     # 1. Let straggler negotiation settle -- read (and respond to DSR)
     #    instead of sleeping blind so early DSR requests get a CPR reply.
     settle_data = await _read_banner_until_quiet(
@@ -629,8 +636,8 @@ async def _fingerprint_session(
             "dsr_replies": cursor.dsr_replies,
         }
     )
-    if writer.mssp_data is not None:
-        session_data["mssp"] = writer.mssp_data
+    if writer.ctx.mssp_data is not None:
+        session_data["mssp"] = writer.ctx.mssp_data
     session_data.update(_collect_mud_data(writer))
 
     session_entry: dict[str, Any] = {
@@ -966,18 +973,18 @@ def _create_server_protocol_fingerprint(
 
 
 def _collect_mud_data(writer: TelnetWriter) -> dict[str, Any]:
-    """Collect MUD protocol data from *writer* into a dict."""
+    """Collect MUD protocol data from *writer.ctx* into a dict."""
     result: dict[str, Any] = {}
-    if writer.zmp_data:
-        result["zmp"] = writer.zmp_data
-    if writer.atcp_data:
-        result["atcp"] = [{"package": pkg, "value": val} for pkg, val in writer.atcp_data]
-    if writer.aardwolf_data:
-        result["aardwolf"] = writer.aardwolf_data
-    if writer.mxp_data:
-        result["mxp"] = [d.hex() if d else "activated" for d in writer.mxp_data]
-    if writer.comport_data:
-        result["comport"] = writer.comport_data
+    if writer.ctx.zmp_data:
+        result["zmp"] = writer.ctx.zmp_data
+    if writer.ctx.atcp_data:
+        result["atcp"] = [{"package": pkg, "value": val} for pkg, val in writer.ctx.atcp_data]
+    if writer.ctx.aardwolf_data:
+        result["aardwolf"] = writer.ctx.aardwolf_data
+    if writer.ctx.mxp_data:
+        result["mxp"] = [d.hex() if d else "activated" for d in writer.ctx.mxp_data]
+    if writer.ctx.comport_data:
+        result["comport"] = writer.ctx.comport_data
     return result
 
 
@@ -1099,10 +1106,10 @@ def _format_banner(data: bytes, encoding: str = "utf-8") -> str:
 
 async def _await_mssp_data(writer: TelnetWriter, deadline: float) -> None:
     """Wait for MSSP data until *deadline* if server acknowledged MSSP."""
-    if not writer.remote_option.enabled(MSSP) or writer.mssp_data is not None:
+    if not writer.remote_option.enabled(MSSP) or writer.ctx.mssp_data is not None:
         return
     remaining = deadline - time.time()
-    while remaining > 0 and writer.mssp_data is None:
+    while remaining > 0 and writer.ctx.mssp_data is None:
         await asyncio.sleep(min(0.05, remaining))
         remaining = deadline - time.time()
 
diff --git a/telnetlib3/stream_writer.py b/telnetlib3/stream_writer.py
index 94f349d8..62f889ac 100644
--- a/telnetlib3/stream_writer.py
+++ b/telnetlib3/stream_writer.py
@@ -16,6 +16,7 @@
 from . import slc
 from .mud import (
     zmp_decode,
+    zmp_encode,
     atcp_decode,
     gmcp_decode,
     gmcp_encode,
@@ -108,6 +109,11 @@
 #: MUD protocol options that a plain telnet client should decline by default.
 _MUD_PROTOCOL_OPTIONS = frozenset({GMCP, MSDP, MSSP, MSP, MXP, ZMP, AARDWOLF, ATCP})
 
+#: Maximum number of bytes buffered between ``IAC SB`` and ``IAC SE``.
+#: A sub-negotiation exceeding this bound is discarded and subsequent
+#: bytes are dropped until the terminating ``IAC SE``.
+_MAX_SUBNEGOTIATION = 1 << 20
+
 
 class TelnetWriter:
     """
@@ -251,6 +257,12 @@ def __init__(
         #: in response to a server WILL (passive negotiation).
         self.passive_do: set[bytes] = set()
 
+        #: Per-option will callbacks invoked after :meth:`handle_will`
+        #: completes standard negotiation.  Keys are option bytes, values
+        #: are lists of ``callable(bytes)``.  Use :meth:`add_will_callback`
+        #: and :meth:`remove_will_callback` to manage.
+        self.will_callbacks: dict[bytes, list[Callable[[bytes], None]]] = {}
+
         #: Whether the encoding was explicitly set (not just the default
         #: ``"ascii"``).  Used by fingerprinting and client connection logic
         #: to decide whether to negotiate CHARSET.
@@ -282,31 +294,6 @@ def __init__(
         #: buffer limit of some telnet clients).
         self._environ_batches: list[list[Union[str, bytes]]] = []
 
-        #: Decoded MSSP variables received via subnegotiation.
-        #: ``None`` until a ``SB MSSP`` payload is received and decoded.
-        self.mssp_data: Optional[dict[str, str | list[str]]] = None
-
-        #: Accumulated ZMP messages (list of [command, arg, ...] lists).
-        #: Empty until ``SB ZMP`` payloads are received and decoded.
-        self.zmp_data: list[list[str]] = []
-
-        #: Accumulated ATCP messages (list of (package, value) tuples).
-        #: Empty until ``SB ATCP`` payloads are received and decoded.
-        self.atcp_data: list[tuple[str, str]] = []
-
-        #: Accumulated Aardwolf messages (list of decoded dicts).
-        #: Empty until ``SB AARDWOLF`` payloads are received and decoded.
-        self.aardwolf_data: list[dict[str, Any]] = []
-
-        #: Accumulated MXP subnegotiation payloads (list of raw bytes).
-        #: Empty until ``SB MXP`` payloads are received.  An empty payload
-        #: (``b""``) signals MXP mode activation.
-        self.mxp_data: list[bytes] = []
-
-        #: COM-PORT-OPTION (RFC 2217) data received via subnegotiation.
-        #: ``None`` until an ``SB COM-PORT-OPTION`` payload is received.
-        self.comport_data: Optional[dict[str, Any]] = None
-
         #: Compression policy: ``None`` = passively accept (default),
         #: ``True`` = actively request, ``False`` = reject.
         self.compression: Optional[bool] = None
@@ -335,6 +322,10 @@ def __init__(
         #: Sub-negotiation buffer
         self._sb_buffer: collections.deque[bytes] = collections.deque()
 
+        #: True when a sub-negotiation exceeded :data:`_MAX_SUBNEGOTIATION`;
+        #: remaining bytes are dropped until the terminating ``IAC SE``.
+        self._sb_overflow = False
+
         #: SLC buffer
         self._slc_buffer: collections.deque[bytes] = collections.deque()
 
@@ -454,6 +445,55 @@ def transport(self) -> Optional[asyncio.BaseTransport]:
         """Return the underlying transport."""
         return self._transport
 
+    # -- Deprecated MUD data properties, delegated to ctx ------------------
+
+    @property
+    def mssp_data(self) -> Optional[dict[str, str | list[str]]]:
+        """Deprecated: use ``writer.ctx.mssp_data``."""
+        return self.ctx.mssp_data
+
+    @mssp_data.setter
+    def mssp_data(self, value: Optional[dict[str, str | list[str]]]) -> None:
+        self.ctx.mssp_data = value
+
+    @property
+    def atcp_data(self) -> list[tuple[str, str]]:
+        """Deprecated: use ``writer.ctx.atcp_data``."""
+        return self.ctx.atcp_data
+
+    @atcp_data.setter
+    def atcp_data(self, value: list[tuple[str, str]]) -> None:
+        self.ctx.atcp_data = value
+
+    @property
+    def aardwolf_data(self) -> list[dict[str, Any]]:
+        """Deprecated: use ``writer.ctx.aardwolf_data``."""
+        return self.ctx.aardwolf_data
+
+    @aardwolf_data.setter
+    def aardwolf_data(self, value: list[dict[str, Any]]) -> None:
+        self.ctx.aardwolf_data = value
+
+    @property
+    def mxp_data(self) -> list[bytes]:
+        """Deprecated: use ``writer.ctx.mxp_data``."""
+        return self.ctx.mxp_data
+
+    @mxp_data.setter
+    def mxp_data(self, value: list[bytes]) -> None:
+        self.ctx.mxp_data = value
+
+    @property
+    def comport_data(self) -> Optional[dict[str, Any]]:
+        """Deprecated: use ``writer.ctx.comport_data``."""
+        return self.ctx.comport_data
+
+    @comport_data.setter
+    def comport_data(self, value: Optional[dict[str, Any]]) -> None:
+        self.ctx.comport_data = value
+
+    # -- end deprecated properties -----------------------------------------
+
     def close(self) -> None:
         """Close the connection and release resources."""
         if self.connection_closed:
@@ -474,6 +514,7 @@ def close(self) -> None:
         self._ext_callback.clear()
         self._ext_send_callback.clear()
         self._ext_offer_callback.clear()
+        self.will_callbacks.clear()
         self._slc_callback.clear()
         self._iac_callback.clear()
         self._protocol = None
@@ -728,8 +769,11 @@ def feed_byte(self, byte: bytes) -> bool:
         if byte == IAC:
             self.iac_received = not self.iac_received
             if not self.iac_received and self.cmd_received == SB:
-                # SB buffer receives escaped IAC values
-                self._sb_buffer.append(IAC)
+                if self._sb_overflow or len(self._sb_buffer) >= _MAX_SUBNEGOTIATION:
+                    self._sb_overflow = True
+                else:
+                    # SB buffer receives escaped IAC values
+                    self._sb_buffer.append(IAC)
 
         elif self.iac_received and not self.cmd_received:
             # parse 2nd byte of IAC
@@ -761,6 +805,15 @@ def feed_byte(self, byte: bytes) -> bool:
                     name_command(cmd),
                 )
                 self._sb_buffer.clear()
+            elif not self._sb_buffer:
+                if self._sb_overflow:
+                    self.log.warning(
+                        "sub-negotiation SB exceeds %d bytes, discarded", _MAX_SUBNEGOTIATION
+                    )
+                else:
+                    self.log.warning(
+                        "sub-negotiation SB with no option byte (IAC SB IAC SE), discarded"
+                    )
             else:
                 # sub-negotiation end (SE), fire handle_subnegotiation
                 self.log.debug(
@@ -772,12 +825,24 @@ def feed_byte(self, byte: bytes) -> bool:
                     self._sb_buffer.clear()
                     self.iac_received = False
             self.iac_received = False
+            self._sb_overflow = False
 
         elif self.cmd_received == SB:
             # continue buffering of sub-negotiation command.
             if not self._sb_buffer:
                 self.log.debug("begin sub-negotiation SB %s", name_command(byte))
-            self._sb_buffer.append(byte)
+            if self._sb_overflow or len(self._sb_buffer) >= _MAX_SUBNEGOTIATION:
+                if not self._sb_overflow:
+                    sb_opt = name_command(self._sb_buffer[0]) if self._sb_buffer else "?"
+                    self.log.warning(
+                        "sub-negotiation SB %s exceeds %d bytes, discarding until IAC SE",
+                        sb_opt,
+                        _MAX_SUBNEGOTIATION,
+                    )
+                self._sb_buffer.clear()
+                self._sb_overflow = True
+            else:
+                self._sb_buffer.append(byte)
 
         elif self.cmd_received:
             # parse 3rd and final byte of IAC DO, DONT, WILL, WONT.
@@ -1084,6 +1149,23 @@ def send_gmcp(self, package: str, data: Any = None) -> None:
         self.log.debug("send IAC SB GMCP %s IAC SE", package)
         self.send_iac(IAC + SB + GMCP + payload + IAC + SE)
 
+    def send_zmp(self, command: str, *args: str) -> None:
+        """
+        Transmit a ZMP message via subnegotiation.
+
+        :param command: ZMP command name (e.g., ``"zmp.ident"``).
+        :param args: Zero or more argument strings.
+        """
+        if not (self.local_option.enabled(ZMP) or self.remote_option.enabled(ZMP)):
+            self.log.debug("cannot send ZMP without negotiation")
+            return
+        payload = self._escape_iac(zmp_encode(command, *args))
+        maybe_args = ""
+        if args:
+            maybe_args = " " + " ".join(args)
+        self.log.debug("send IAC SB ZMP %s%s IAC SE", command, maybe_args)
+        self.send_iac(IAC + SB + ZMP + payload + IAC + SE)
+
     def send_msdp(self, variables: dict[str, Any]) -> None:
         """
         Transmit MSDP variables via subnegotiation.
@@ -1705,6 +1787,35 @@ def set_ext_callback(self, cmd: bytes, func: Callable[..., Any]) -> None:
         """
         self._ext_callback[cmd] = func
 
+    def add_will_callback(self, opt: bytes, func: Callable[[bytes], None]) -> None:
+        """
+        Register *func* to be called after :meth:`handle_will` processes *opt*.
+
+        Multiple callbacks may be registered for the same option.  They are
+        invoked in registration order after the standard negotiation logic
+        in :meth:`handle_will` completes.
+
+        :param opt: Telnet option byte (e.g. ``GMCP``, ``ZMP``, ``CHARSET``).
+        :param func: Callable receiving the option byte ``opt``.
+        """
+        self.will_callbacks.setdefault(opt, []).append(func)
+
+    def remove_will_callback(self, opt: bytes, func: Callable[[bytes], None]) -> None:
+        """
+        Remove a previously registered will callback for *opt*.
+
+        :param opt: Telnet option byte.
+        :param func: The exact callable previously passed to
+            :meth:`add_will_callback`.
+        :raises ValueError: If *func* is not registered for *opt*.
+        """
+        try:
+            self.will_callbacks[opt].remove(func)
+        except (KeyError, ValueError):
+            raise ValueError(f"{func} not registered for option {opt!r}") from None
+        if not self.will_callbacks[opt]:
+            del self.will_callbacks[opt]
+
     def handle_xdisploc(self, xdisploc: str) -> None:
         """Receive XDISPLAY value ``xdisploc``, :rfc:`1096`."""
         #   xdisploc string format is '<host>:<dispnum>[.<screennum>]'.
@@ -1787,7 +1898,7 @@ def handle_gmcp(self, package: str, data: Any) -> None:
         Receive GMCP message with ``package`` name and ``data``.
 
         :param package: GMCP package name (e.g., ``"Char.Vitals"``).
-        :param data: Decoded JSON value -- may be any JSON type
+        :param data: Decoded JSON value, can be any JSON type
             (``str``, ``int``, ``float``, ``bool``, ``None``,
             ``list``, or ``dict``).
         """
@@ -1817,10 +1928,10 @@ def handle_mxp(self, data: bytes) -> None:
         self.log.debug("MXP: %r", data)
         self.mxp_data.append(data)
 
-    def handle_zmp(self, parts: list[str]) -> None:
-        """Receive decoded ZMP message as list of ``[command, arg, ...]``."""
-        self.log.debug("ZMP: %r", parts)
-        self.zmp_data.append(parts)
+    def handle_zmp(self, command: str, *args: str) -> None:
+        """Receive decoded ZMP message as ``command`` and ``*args``."""
+        self.log.debug("ZMP: %s %r", command, args)
+        self.ctx.zmp_data[command] = list(args)
 
     def handle_aardwolf(self, data: dict[str, Any]) -> None:
         """Receive decoded Aardwolf message as dict."""
@@ -1967,7 +2078,10 @@ def handle_do(self, opt: bytes) -> bool:
                     if not self.local_option.enabled(opt):
                         self.iac(WILL, opt)
                     return True
-                self.log.debug("DO %s: MUD protocol, declining on client.", name_command(opt))
+                self.log.debug(
+                    "DO %s: MUD protocol, declining on client (enable with always_will).",
+                    name_command(opt),
+                )
                 if not self.local_option.enabled(opt):
                     self.iac(WONT, opt)
                 return False
@@ -2096,7 +2210,14 @@ def handle_will(self, opt: bytes) -> None:
                     if not self.remote_option.enabled(opt):
                         self.iac(DO, opt)
                         self.remote_option[opt] = True
+                    for callback in self.will_callbacks.get(opt, ()):
+                        callback(opt)
                     return
+                self.log.debug(
+                    "WILL %s: MUD protocol, declining on client "
+                    "(enable with always_do or passive_do).",
+                    name_command(opt),
+                )
                 self.iac(DONT, opt)
                 return
             # Reject MCCP when compression is disabled or TLS is active
@@ -2198,6 +2319,9 @@ def handle_will(self, opt: bytes) -> None:
             if self.pending_option.enabled(DO + opt):
                 self.pending_option[DO + opt] = False
 
+        for callback in self.will_callbacks.get(opt, ()):
+            callback(opt)
+
     def handle_wont(self, opt: bytes) -> None:
         """
         Process byte 3 of series (IAC, WONT, opt) received by remote end.
@@ -3199,7 +3323,8 @@ def _handle_sb_zmp(self, buf: collections.deque[bytes]) -> None:
         payload = b"".join(buf)
         encoding = self.environ_encoding or "utf-8"
         parts = zmp_decode(payload, encoding=encoding)
-        self._ext_callback[ZMP](parts)
+        if parts:
+            self._ext_callback[ZMP](*parts)
 
     def _handle_sb_aardwolf(self, buf: collections.deque[bytes]) -> None:
         """
diff --git a/telnetlib3/tests/test_client_unit.py b/telnetlib3/tests/test_client_unit.py
index 30822608..03935878 100644
--- a/telnetlib3/tests/test_client_unit.py
+++ b/telnetlib3/tests/test_client_unit.py
@@ -2,6 +2,7 @@
 import sys
 import types
 import asyncio
+import logging
 from unittest import mock
 
 # 3rd party
@@ -517,10 +518,75 @@ async def test_on_gmcp_merges_dicts_on_writer_ctx():
     assert client.writer.ctx.gmcp_data["Char.Vitals"] == {"hp": 63, "maxhp": 100}
 
 
-def test_default_gmcp_modules_are_lowercase():
-    for spec in cl._DEFAULT_GMCP_MODULES:
-        module_part = spec.rsplit(" ", 1)[0]
-        assert module_part == module_part.lower()
+@pytest.mark.asyncio
+async def test_on_zmp_stores_on_ctx_zmp_data():
+    client, _ = _make_connected_client()
+    client.on_zmp("char.vitals", "hp", "100")
+    assert client.writer.ctx.zmp_data == {"char.vitals": ["hp", "100"]}
+
+
+@pytest.mark.asyncio
+async def test_on_zmp_check_support():
+    client, transport = _make_connected_client(zmp_check_handler=lambda cmd: True)
+    from telnetlib3.telopt import ZMP
+
+    client.writer.remote_option[ZMP] = True
+    client._zmp_ident_sent = True
+    client.on_zmp("zmp.check", "char.vitals")
+    assert client.writer.ctx.zmp_data == {"zmp.check": ["char.vitals"]}
+    sent = bytes(transport.data)
+    assert b"zmp.support\x00char.vitals\x00" in sent
+
+
+@pytest.mark.asyncio
+async def test_on_zmp_check_no_support():
+    client, transport = _make_connected_client(zmp_check_handler=lambda cmd: False)
+    from telnetlib3.telopt import ZMP
+
+    client.writer.remote_option[ZMP] = True
+    client._zmp_ident_sent = True
+    client.on_zmp("zmp.check", "char.vitals")
+    sent = bytes(transport.data)
+    assert b"zmp.no-support\x00char.vitals\x00" in sent
+
+
+@pytest.mark.asyncio
+async def test_on_zmp_check_default_refuses():
+    client, transport = _make_connected_client()
+    from telnetlib3.telopt import ZMP
+
+    client.writer.remote_option[ZMP] = True
+    client._zmp_ident_sent = True
+    client.on_zmp("zmp.check", "char.vitals")
+    sent = bytes(transport.data)
+    assert b"zmp.no-support\x00char.vitals\x00" in sent
+
+
+@pytest.mark.asyncio
+async def test_zmp_ident_sent_on_will_zmp():
+    client, transport = _make_connected_client(zmp_check_handler=lambda cmd: True)
+    from telnetlib3.telopt import ZMP
+
+    client.writer.handle_will(ZMP)
+    sent = bytes(transport.data)
+    assert b"zmp.ident\x00telnetlib3\x00" in sent
+
+
+@pytest.mark.asyncio
+async def test_zmp_send_support_responds_all():
+    """zmp.send-support with no args responds with all supported commands."""
+    client, transport = _make_connected_client(
+        zmp_check_handler=lambda cmd: True, zmp_supported_commands={"char.vitals", "room.info"}
+    )
+    from telnetlib3.telopt import ZMP
+
+    client.writer.remote_option[ZMP] = True
+    # Ident not yet sent -- send-support should trigger a full response.
+    client._zmp_ident_sent = False
+    client.on_zmp("zmp.send-support")
+    sent = bytes(transport.data)
+    assert b"zmp.support\x00char.vitals\x00" in sent
+    assert b"zmp.support\x00room.info\x00" in sent
 
 
 @pytest.mark.asyncio
@@ -530,7 +596,7 @@ async def test_send_gmcp_hello_lowercases_default_modules():
     mock_writer = mock.Mock()
     mock_writer.send_gmcp.side_effect = lambda pkg, data: calls.append((pkg, data))
     client.writer = mock_writer
-    client._send_gmcp_hello()
+    client.send_gmcp_hello()
     assert client._gmcp_hello_sent is True
     assert len(calls) == 2
     pkg_name, supports_set = calls[1]
@@ -546,7 +612,7 @@ async def test_send_gmcp_hello_lowercases_consumer_modules():
     mock_writer = mock.Mock()
     mock_writer.send_gmcp.side_effect = lambda pkg, data: calls.append((pkg, data))
     client.writer = mock_writer
-    client._send_gmcp_hello()
+    client.send_gmcp_hello()
     _, supports_set = calls[1]
     assert "room.info 1" in supports_set
     assert "char 1" in supports_set
@@ -557,9 +623,9 @@ async def test_send_gmcp_hello_idempotent():
     client = _make_client()
     mock_writer = mock.Mock()
     client.writer = mock_writer
-    client._send_gmcp_hello()
+    client.send_gmcp_hello()
     call_count = mock_writer.send_gmcp.call_count
-    client._send_gmcp_hello()
+    client.send_gmcp_hello()
     assert mock_writer.send_gmcp.call_count == call_count
 
 
@@ -571,3 +637,65 @@ async def _bad_fp():
     with pytest.raises(SystemExit) as exc_info:
         cl.fingerprint_main()
     assert exc_info.value.code == 1
+
+
+@pytest.mark.asyncio
+async def test_on_will_gmcp_sends_hello():
+    client, transport = _make_connected_client()
+    from telnetlib3.telopt import GMCP
+
+    client.writer.remote_option[GMCP] = True
+    client.on_will_gmcp(GMCP)
+    sent = bytes(transport.data)
+    assert b"Core.Hello" in sent
+
+
+@pytest.mark.asyncio
+async def test_on_will_gmcp_noop_when_not_enabled():
+    client, transport = _make_connected_client()
+    from telnetlib3.telopt import GMCP
+
+    client.on_will_gmcp(GMCP)
+    sent = bytes(transport.data)
+    assert b"Core.Hello" not in sent
+
+
+@pytest.mark.asyncio
+async def test_on_will_zmp_sends_ident():
+    client, transport = _make_connected_client(zmp_check_handler=lambda cmd: True)
+    from telnetlib3.telopt import ZMP
+
+    client.writer.remote_option[ZMP] = True
+    client.on_will_zmp(ZMP)
+    sent = bytes(transport.data)
+    assert b"zmp.ident\x00telnetlib3\x00" in sent
+
+
+@pytest.mark.asyncio
+async def test_on_will_charset_logs_when_both_sides_enabled(caplog):
+    client, _ = _make_connected_client()
+    from telnetlib3.telopt import CHARSET
+
+    client.writer.remote_option[CHARSET] = True
+    client.writer.local_option[CHARSET] = True
+    with caplog.at_level(logging.DEBUG):
+        client.on_will_charset(CHARSET)
+    assert "Both sides support CHARSET" in caplog.text
+
+
+@pytest.mark.asyncio
+async def test_setup_gmcp_registers_will_callback():
+    client, _ = _make_connected_client()
+    from telnetlib3.telopt import GMCP
+
+    on_will_gmcp = client.on_will_gmcp
+    assert on_will_gmcp in client.writer.will_callbacks.get(GMCP, [])
+
+
+@pytest.mark.asyncio
+async def test_setup_zmp_registers_will_callback():
+    client, _ = _make_connected_client()
+    from telnetlib3.telopt import ZMP
+
+    on_will_zmp = client.on_will_zmp
+    assert on_will_zmp in client.writer.will_callbacks.get(ZMP, [])
diff --git a/telnetlib3/tests/test_mccp.py b/telnetlib3/tests/test_mccp.py
index 3f32b484..8b9b75ec 100644
--- a/telnetlib3/tests/test_mccp.py
+++ b/telnetlib3/tests/test_mccp.py
@@ -771,3 +771,68 @@ async def test_raw_deflate_mid_chunk(self):
         joined = b"".join(received)
         assert joined == plaintext
         assert client._mccp2_wbits_fallback is True
+
+
+@pytest.mark.asyncio
+class TestMCCPTraceLogging:
+    def _recv_messages(self, caplog):
+        return [r.getMessage() for r in caplog.records if r.getMessage().startswith("recv ")]
+
+    async def test_client_trace_shows_decompressed_not_compressed(self, caplog):
+        """TRACE recv logging shows the decompressed telnet stream, never zlib bytes."""
+        from telnetlib3.accessories import TRACE
+
+        client, received = _make_client_with_capture()
+        plaintext = b'room.info {"visibility":0} '
+        compressor = zlib.compressobj(
+            zlib.Z_BEST_COMPRESSION, zlib.DEFLATED, 12, 5, zlib.Z_DEFAULT_STRATEGY
+        )
+        first = compressor.compress(plaintext) + compressor.flush(zlib.Z_SYNC_FLUSH)
+        second = compressor.compress(b"more plaintext") + compressor.flush(zlib.Z_SYNC_FLUSH)
+
+        old_level = client.log.level
+        client.log.setLevel(TRACE)
+        try:
+            with caplog.at_level(TRACE):
+                client._process_chunk(_BOUNDARY_SB + first)
+                client._process_chunk(second)
+        finally:
+            client.log.setLevel(old_level)
+
+        assert b"".join(received) == plaintext + b"more plaintext"
+        messages = self._recv_messages(caplog)
+        assert messages
+        # The chunk activating MCCP2 dumps only its plain IAC SB prefix raw.
+        assert any(m.startswith("recv 5 bytes") for m in messages)
+        decompressed = [m for m in messages if "(decompressed)" in m]
+        assert decompressed
+        assert any('room.info {"visi' in m for m in decompressed)
+        # The zlib bytes never appear in any dump.
+        compressed_hex = first[:16].hex(" ")
+        assert all(compressed_hex not in m for m in messages)
+
+    async def test_server_trace_shows_decompressed_not_compressed(self, caplog):
+        """Server TRACE recv logging shows decompressed client→server data."""
+        from telnetlib3 import server_base as server_base_module
+        from telnetlib3.accessories import TRACE
+        from telnetlib3.server_base import BaseServer
+
+        server = BaseServer(encoding=False, connect_maxwait=0.1)
+        transport = MockTransport()
+        server.connection_made(transport)
+        server._mccp3_decompressor = zlib.decompressobj()
+
+        plaintext = b"hello from compressed client"
+        compressed = _make_compressed(plaintext)
+        old_level = server_base_module.logger.level
+        server_base_module.logger.setLevel(TRACE)
+        try:
+            with caplog.at_level(TRACE):
+                server.data_received(compressed)
+        finally:
+            server_base_module.logger.setLevel(old_level)
+
+        messages = self._recv_messages(caplog)
+        assert messages
+        assert any("(decompressed)" in m and "hello from compr" in m for m in messages)
+        assert all(compressed[:16].hex(" ") not in m for m in messages)
diff --git a/telnetlib3/tests/test_mud.py b/telnetlib3/tests/test_mud.py
index 7aa06a1d..daa204e2 100644
--- a/telnetlib3/tests/test_mud.py
+++ b/telnetlib3/tests/test_mud.py
@@ -6,6 +6,7 @@
 # local
 from telnetlib3.mud import (
     zmp_decode,
+    zmp_encode,
     atcp_decode,
     gmcp_decode,
     gmcp_encode,
@@ -348,3 +349,28 @@ def test_mssp_decode_skips_garbage_bytes():
     buf = b"\x42" + MSSP_VAR + b"NAME" + MSSP_VAL + b"TestMUD"
     result = mssp_decode(buf)
     assert result == {"NAME": "TestMUD"}
+
+
+def test_zmp_encode_decode_roundtrip():
+    encoded = zmp_encode("zmp.ident", "telnetlib3", "1.0")
+    assert encoded == b"zmp.ident\x00telnetlib3\x001.0\x00"
+    decoded = zmp_decode(encoded)
+    assert decoded == ["zmp.ident", "telnetlib3", "1.0"]
+
+
+def test_zmp_encode_no_args():
+    encoded = zmp_encode("zmp.ping")
+    assert encoded == b"zmp.ping\x00"
+    assert zmp_decode(encoded) == ["zmp.ping"]
+
+
+def test_zmp_encode_multiple_args():
+    encoded = zmp_encode("zmp.check", "char.vitals")
+    assert encoded == b"zmp.check\x00char.vitals\x00"
+    assert zmp_decode(encoded) == ["zmp.check", "char.vitals"]
+
+
+def test_zmp_decode_latin1_fallback():
+    buf = b"echo\x00caf\xe9\x00"
+    result = zmp_decode(buf)
+    assert result == ["echo", "caf\xe9"]
diff --git a/telnetlib3/tests/test_mud_negotiation.py b/telnetlib3/tests/test_mud_negotiation.py
index 1f918525..7fea8b8d 100644
--- a/telnetlib3/tests/test_mud_negotiation.py
+++ b/telnetlib3/tests/test_mud_negotiation.py
@@ -193,6 +193,20 @@ def test_send_gmcp_not_negotiated():
     assert len(t.writes) == 0
 
 
+def test_send_zmp():
+    w, t, p = new_writer(server=True)
+    w.local_option[ZMP] = True
+    w.send_zmp("zmp.ident", "MudName", "1.0")
+    expected = IAC + SB + ZMP + b"zmp.ident\x00MudName\x001.0\x00" + IAC + SE
+    assert expected in t.writes
+
+
+def test_send_zmp_not_negotiated():
+    w, t, p = new_writer(server=True)
+    w.send_zmp("zmp.ident", "MudName", "1.0")
+    assert len(t.writes) == 0
+
+
 def test_send_msdp():
     w, t, p = new_writer(server=True)
     w.local_option[MSDP] = True
@@ -356,7 +370,7 @@ def test_sb_zmp_dispatch():
     payload = b"zmp.ident\x00MudName\x001.0\x00A test MUD\x00"
     buf = collections.deque([bytes([ZMP[0]])] + [bytes([b]) for b in payload])
     w.handle_subnegotiation(buf)
-    assert w.zmp_data == [["zmp.ident", "MudName", "1.0", "A test MUD"]]
+    assert w.ctx.zmp_data == {"zmp.ident": ["MudName", "1.0", "A test MUD"]}
 
 
 def test_sb_zmp_empty_payload():
@@ -364,20 +378,18 @@ def test_sb_zmp_empty_payload():
     w.pending_option[SB + ZMP] = True
     buf = collections.deque([bytes([ZMP[0]])])
     w.handle_subnegotiation(buf)
-    assert w.zmp_data == [[]]
+    assert w.ctx.zmp_data == {}
 
 
-def test_sb_zmp_accumulates():
+def test_sb_zmp_replaces():
     w, _t, _p = new_writer(server=True)
     w.pending_option[SB + ZMP] = True
-    buf1 = collections.deque([bytes([ZMP[0]])] + [bytes([b]) for b in b"zmp.ping\x00"])
+    buf1 = collections.deque([bytes([ZMP[0]])] + [bytes([b]) for b in b"char.vitals\x00hp=100\x00"])
     w.handle_subnegotiation(buf1)
     w.pending_option[SB + ZMP] = True
-    buf2 = collections.deque([bytes([ZMP[0]])] + [bytes([b]) for b in b"zmp.check\x00zmp.ping\x00"])
+    buf2 = collections.deque([bytes([ZMP[0]])] + [bytes([b]) for b in b"char.vitals\x00hp=50\x00"])
     w.handle_subnegotiation(buf2)
-    assert len(w.zmp_data) == 2
-    assert w.zmp_data[0] == ["zmp.ping"]
-    assert w.zmp_data[1] == ["zmp.check", "zmp.ping"]
+    assert w.ctx.zmp_data == {"char.vitals": ["hp=50"]}
 
 
 def test_sb_atcp_dispatch():
diff --git a/telnetlib3/tests/test_server_fingerprinting.py b/telnetlib3/tests/test_server_fingerprinting.py
index fec6f2ae..a1790872 100644
--- a/telnetlib3/tests/test_server_fingerprinting.py
+++ b/telnetlib3/tests/test_server_fingerprinting.py
@@ -9,6 +9,7 @@
 
 # local
 from telnetlib3 import fingerprinting as fps
+from telnetlib3 import _session_context
 from telnetlib3 import server_fingerprinting as sfp
 from telnetlib3.telopt import VAR, USERVAR
 
@@ -46,12 +47,8 @@ def __init__(self, extra=None, will_options=None, wont_options=None):
         self.local_option = MockOption()
         self.environ_encoding = "ascii"
         self.environ_send_raw = None
-        self.mssp_data = None
-        self.zmp_data: list[list[str]] = []
-        self.atcp_data: list[tuple[str, str]] = []
-        self.aardwolf_data: list[dict[str, object]] = []
-        self.mxp_data: list[bytes] = []
-        self.comport_data: dict[str, object] | None = None
+        self.passive_do: set[bytes] = set()
+        self.ctx = _session_context.TelnetSessionContext()
         self.protocol = _MockProtocol()
         self._closing = False
         self._menu_inline: bool = False
@@ -88,6 +85,46 @@ def is_closing(self):
     def close(self):
         self._closing = True
 
+    @property
+    def mssp_data(self):
+        return self.ctx.mssp_data
+
+    @mssp_data.setter
+    def mssp_data(self, value):
+        self.ctx.mssp_data = value
+
+    @property
+    def atcp_data(self):
+        return self.ctx.atcp_data
+
+    @atcp_data.setter
+    def atcp_data(self, value):
+        self.ctx.atcp_data = value
+
+    @property
+    def aardwolf_data(self):
+        return self.ctx.aardwolf_data
+
+    @aardwolf_data.setter
+    def aardwolf_data(self, value):
+        self.ctx.aardwolf_data = value
+
+    @property
+    def mxp_data(self):
+        return self.ctx.mxp_data
+
+    @mxp_data.setter
+    def mxp_data(self, value):
+        self.ctx.mxp_data = value
+
+    @property
+    def comport_data(self):
+        return self.ctx.comport_data
+
+    @comport_data.setter
+    def comport_data(self, value):
+        self.ctx.comport_data = value
+
 
 class MockReader:
     def __init__(self, chunks=None):
@@ -1300,7 +1337,7 @@ async def test_banner_loop_no_prompt_detected(tmp_path):
 async def test_session_data_mud_protocol_fields(tmp_path):
     """Session data includes MUD protocol fields when present."""
     writer = MockWriter(will_options=[fps.SGA])
-    writer.zmp_data = [["check", "telnetlib3"]]
+    writer.ctx.zmp_data = {"check": ["telnetlib3"]}
     writer.atcp_data = [("Auth.Request", "ON")]
     writer.aardwolf_data = [{"type": "stats"}]
     writer.mxp_data = [None, b"\x01\x02"]
@@ -1311,7 +1348,7 @@ async def test_session_data_mud_protocol_fields(tmp_path):
     with open(save_path, encoding="utf-8") as f:
         data = json.load(f)
     session = data["server-probe"]["session_data"]
-    assert session["zmp"] == [["check", "telnetlib3"]]
+    assert session["zmp"] == {"check": ["telnetlib3"]}
     assert session["atcp"] == [{"package": "Auth.Request", "value": "ON"}]
     assert session["aardwolf"] == [{"type": "stats"}]
     assert session["mxp"] == ["activated", "0102"]
diff --git a/telnetlib3/tests/test_stream_writer_full.py b/telnetlib3/tests/test_stream_writer_full.py
index 4ea409c0..c79104ff 100644
--- a/telnetlib3/tests/test_stream_writer_full.py
+++ b/telnetlib3/tests/test_stream_writer_full.py
@@ -1723,3 +1723,60 @@ def test_handle_will_client_directional_refusal():
     w.handle_will(TTYPE)
     assert t.writes[-1] == IAC + DONT + TTYPE
     assert TTYPE in w.directional_refusals
+
+
+def test_add_will_callback_fires_on_matching_opt():
+    w, t, _ = new_writer(server=True)
+    called: list[bytes] = []
+    w.add_will_callback(GMCP, lambda opt: called.append(opt))
+    w.handle_will(GMCP)
+    assert called == [GMCP]
+
+
+def test_add_will_callback_does_not_fire_on_nonmatching_opt():
+    w, t, _ = new_writer(server=False, client=True)
+    called: list[bytes] = []
+    w.add_will_callback(GMCP, lambda opt: called.append(opt))
+    w.handle_will(NAWS)
+    assert called == []
+
+
+def test_add_will_callback_multiple_for_same_opt():
+    w, t, _ = new_writer(server=True)
+    called: list[int] = []
+    w.add_will_callback(GMCP, lambda opt: called.append(1))
+    w.add_will_callback(GMCP, lambda opt: called.append(2))
+    w.handle_will(GMCP)
+    assert called == [1, 2]
+
+
+def test_remove_will_callback():
+    w, t, _ = new_writer(server=True)
+    called: list[bytes] = []
+
+    def _cb(opt: bytes) -> None:
+        called.append(opt)
+
+    w.add_will_callback(GMCP, _cb)
+    w.remove_will_callback(GMCP, _cb)
+    w.handle_will(GMCP)
+    assert called == []
+
+
+def test_remove_will_callback_raises_if_not_registered():
+    w, t, _ = new_writer(server=True)
+
+    def _cb(opt: bytes) -> None:
+        pass
+
+    with pytest.raises(ValueError):
+        w.remove_will_callback(GMCP, _cb)
+
+
+def test_close_clears_will_callbacks():
+    w, t, _ = new_writer(server=True)
+    called: list[bytes] = []
+    w.add_will_callback(GMCP, lambda opt: called.append(opt))
+    w.close()
+    assert w.will_callbacks == {}
+    assert not called
diff --git a/telnetlib3/tests/test_writer.py b/telnetlib3/tests/test_writer.py
index 48a5dfd7..5d4836c9 100644
--- a/telnetlib3/tests/test_writer.py
+++ b/telnetlib3/tests/test_writer.py
@@ -151,6 +151,47 @@ def test_sb_interrupted():
     assert writer.feed_byte(SE) is True
 
 
+def test_sb_empty_subnegotiation():
+    """IAC SB IAC SE (no option byte) is discarded without raising."""
+    writer = telnetlib3.TelnetWriter(transport=None, protocol=None, server=True)
+
+    given = IAC + SB + IAC + SE + b"ok"
+    for val in given:
+        writer.feed_byte(bytes([val]))
+
+    assert b"".join(writer._sb_buffer) == b""
+    assert writer.iac_received is False
+    assert writer.cmd_received is False
+    assert writer._sb_overflow is False
+
+
+def test_sb_overflow_discarded():
+    """SB payload exceeding the size bound is discarded, staying in sync."""
+    from telnetlib3.stream_writer import _MAX_SUBNEGOTIATION
+
+    writer = telnetlib3.TelnetWriter(transport=None, protocol=None, server=True)
+    writer.feed_byte(IAC)
+    writer.feed_byte(SB)
+    writer.feed_byte(b"g")
+    writer._sb_buffer.extend([b"\x00"] * (_MAX_SUBNEGOTIATION - 1))
+
+    # one more byte crosses the bound: buffered data is discarded
+    writer.feed_byte(b"\x00")
+    assert writer._sb_overflow is True
+    assert b"".join(writer._sb_buffer) == b""
+
+    # bytes (including escaped IAC) are dropped while overflowing
+    writer.feed_byte(b"\x00")
+    writer.feed_byte(IAC)
+    writer.feed_byte(IAC)
+
+    # terminating SE ends the discard; stream stays in sync
+    writer.feed_byte(IAC)
+    writer.feed_byte(SE)
+    assert writer._sb_overflow is False
+    assert writer.feed_byte(b"z") is True
+
+
 async def test_iac_do_twice_replies_once(bind_host, unused_tcp_port):
     """WILL/WONT replied only once for repeated DO."""