wshallwshall · Oct 17, 2026 · Oct 17, 2026
@@ -308,6 +308,15 @@ that list; see its amendment below.
    record of what was found.
 2. **An alphanumeric MSH-1 gets an unreadable ACK.** The message is accepted, and the ACK echoes the
    letter separator, so MSA-1 (itself letters) cannot be read back.
+
+   **Amendment 2026-10-16, defect 2 FIXED by BACKLOG #2875.** `build_ack` in
+   `messagefoundry/mllpcodec.py` now answers an inbound whose MSH-1 or MSH-2 holds a character of
+   the ACK's own fixed text, or whose delimiters share a character, under `|^~\&`, and the vendored
+   `tee/mllp.py` does the same. A letter the ACK never writes, such as `D`, is still echoed, because
+   the ACK reads back under it; `_ack_echo` holds the rule and what it does not cover. The strict
+   xfail became a plain test. The `alphanumeric-field-separator` policy entry and its discriminator
+   came out in the same change, so `known_defects` is empty and the run tolerates nothing. The
+   `letter-field-separator` catalogue case stays on as a regression case.
 3. **The raw-TCP and X12 listeners have no frame deadline.** A peer trickling inside
    `receive_timeout` holds its slot indefinitely. `max_frame_seconds` is MLLP-only. This one is kept
    out of the run's catalogue and lives only as a strict xfail.

@@ -11,7 +11,8 @@
 across reads or pack several into one.
 
 ACKs are built from the inbound MSH (echoing its encoding characters, swapping
-sender/receiver, copying the original control id into MSA-2). ``ack_mode`` selects the
+sender/receiver, copying the original control id into MSA-2), except where those encoding
+characters collide with the ACK's own text (BACKLOG #2875; ``_ack_echo``). ``ack_mode`` selects the
 MSA-1 code family: ``original`` → AA/AE/AR, ``enhanced`` → CA/CE/CR.
 
 **Why this module exists (BACKLOG #1697).** A client may not import the engine's runtime packages
@@ -35,6 +36,7 @@
 from __future__ import annotations
 
 import logging
+import string
 from enum import Enum
 from typing import TYPE_CHECKING
 
@@ -158,30 +160,87 @@ def _escape_ack_text(text: str, *, field_sep: str, enc: str) -> str:
     return text.replace(esc, f"{esc}E{esc}").replace(field_sep, f"{esc}F{esc}")
 
 
+def _roles(field_sep: str, enc: str) -> dict[str, str]:
+    """Each delimiter character to its role, in MSH-2 order, a short MSH-2 filled from the defaults
+    as the parser fills it. Where two roles share a character the first wins; one with no role is
+    data."""
+    roles: dict[str, str] = {}
+    enc = enc[:4] + _DEFAULT_ENC[len(enc[:4]) :]
+    for char, role in zip(field_sep + enc, "FSRET", strict=True):
+        roles.setdefault(char, role)
+    return roles
+
+
+_DEFAULT_ROLES = _roles(_DEFAULT_FIELD_SEP, _DEFAULT_ENC)
+
 _CODES = {
     AckMode.ORIGINAL: {"AA": "AA", "AE": "AE", "AR": "AR"},
     AckMode.ENHANCED: {"AA": "CA", "AE": "CE", "AR": "CR"},
 }
 
+#: The characters of the ACK's fixed text: the segment ids, ``ACK``, ``P``, every MSA-1 code, the
+#: letters of the ``\E\`` and ``\F\`` escapes :func:`_escape_ack_text` writes, the default MSH-7
+#: (digits and a ``+``/``-`` offset) and the default version ``2.5.1``. An inbound delimiter from
+#: this set would split that text when the ACK is read back, so the ACK is built under the defaults
+#: instead (BACKLOG #2875). Not covered: a caller's own ``timestamp`` or ``control_id``, and the
+#: rest of MSA-3's free text, of which only the escape character and field separator are escaped.
+_ACK_LITERAL_CHARS = frozenset(
+    "MSH"
+    + "MSA"
+    + "ACK"
+    + "P"
+    + "EF"
+    + "".join(_CODES[AckMode.ENHANCED].values())
+    + "".join(_CODES[AckMode.ORIGINAL].values())
+    + string.digits
+    + "+-."
+)
+
 
 def _ack_echo(peek: Peek | None) -> tuple[str, str, str, str, str, str, str, str]:
     """The inbound header values an ACK echoes, or their defaults when there is no ``peek``.
 
     Returns ``(field_sep, enc, sending_app, sending_fac, receiving_app, receiving_fac, version,
     control_id)``. Every value is echoed from the (untrusted) inbound message, so CR/LF is stripped to
     prevent segment injection into the ACK; MSA-3 free text is escaped separately (HL7-3).
+
+    The inbound's MSH-1 and MSH-2 are echoed unless one of them holds a character of the ACK's
+    fixed text (:data:`_ACK_LITERAL_CHARS`), or two delimiters share a character. Then the defaults
+    ``|^~\\&`` are returned instead, so the ACK reads back as an ACK (BACKLOG #2875). An echoed
+    value is then kept only where it holds neither set's escape character and each default delimiter
+    in it has the same role in the inbound (``C^1`` under MSH-2 ``^AB``); any other is blanked.
+    Blanked, not escaped: the value is raw text split under the inbound's delimiters, so escaping it
+    would flatten its structure and re-escape its own escapes (``\\F\\`` names the field separator of
+    whichever set reads it), and an escaped control id would no longer match the one the store
+    recorded. Under the inbound's own delimiters the values need neither, because they were split
+    under them.
     """
     if peek is None:
         return _DEFAULT_FIELD_SEP, _DEFAULT_ENC, "", "", "", "", "2.5.1", ""
+    field_sep = peek.field("MSH-1") or _DEFAULT_FIELD_SEP
+    enc = peek.field("MSH-2") or _DEFAULT_ENC
+    # The characters an echoed value may not hold; none unless the ACK falls back to the defaults.
+    unsafe: frozenset[str] = frozenset()
+    roles = _roles(field_sep, enc)
+    # Fewer than five roles means two share a character, the parser's default-filled slots included.
+    if len(roles) < 5 or not _ACK_LITERAL_CHARS.isdisjoint(field_sep + enc):
+        unsafe = frozenset(c for c, role in _DEFAULT_ROLES.items() if roles.get(c) != role)
+        unsafe |= {_DEFAULT_ENC[2], enc[2:3] or _DEFAULT_ENC[2]}
+        field_sep, enc = _DEFAULT_FIELD_SEP, _DEFAULT_ENC
+
+    def echo(value: str | None, default: str = "") -> str:
+        value = _no_seg_sep(value or "")
+        return (value if unsafe.isdisjoint(value) else "") or default
+
     return (
-        peek.field("MSH-1") or _DEFAULT_FIELD_SEP,
-        peek.field("MSH-2") or _DEFAULT_ENC,
-        _no_seg_sep(peek.sending_app or ""),
-        _no_seg_sep(peek.sending_facility or ""),
-        _no_seg_sep(peek.receiving_app or ""),
-        _no_seg_sep(peek.receiving_facility or ""),
-        _no_seg_sep(peek.version or "2.5.1"),
-        _no_seg_sep(peek.control_id or ""),
+        field_sep,
+        enc,
+        echo(peek.sending_app),
+        echo(peek.sending_facility),
+        echo(peek.receiving_app),
+        echo(peek.receiving_facility),
+        echo(peek.version, "2.5.1"),
+        echo(peek.control_id),
     )
 
 
@@ -208,6 +267,11 @@ def build_ack(
     the same wall-clock hour occurs twice — so a receiver correlating an acknowledgement against its
     own UTC-stamped record would mis-order events by an hour. An explicit offset pins the instant.
     An operator-supplied ``timestamp`` is used verbatim; the caller owns its form.
+
+    The ACK echoes the inbound's MSH-1 and MSH-2, except where one of them holds a character of the
+    ACK's fixed text; then the ACK, and the NAK this also builds, goes out under ``|^~\\&``
+    (BACKLOG #2875; :func:`_ack_echo` holds the rule). MSA-3's escape character and field separator
+    are escaped under whichever set is used.
     """
     # Imported here, not at module top, so a framing-only client never loads the parsing library.
     # It was also a cycle once: `parsing` reached `config.models`, which imports AckMode from here.

@@ -47,9 +47,7 @@
     "min_liveness_probes": 50,
     "min_log_records_seen": 1
   },
-  "known_defects": {
-    "alphanumeric-field-separator": "Engine defect found by this pass on 2026-09-26: a message whose MSH-1 is a letter or digit is accepted, and the ACK echoes that separator, so MSA-1 (AA/AE/AR, themselves letters) cannot be read back by the sender. Pinned by a strict xfail in tests/test_dast_ingress_sweep.py."
-  },
+  "known_defects": {},
   "resource_bounds": {
     "passes": 2,
     "max_heap_growth_bytes": 4194304,

@@ -312,19 +312,14 @@ def reference_frames(plane: str, data: bytes, cap: int) -> tuple[list[bytes], bo
     return frames, False
 
 
-def _alphanumeric_field_separator(payload: bytes) -> bool:
-    return payload[:3] == b"MSH" and payload[3:4].isalnum()
-
-
 #: Known engine defects, each recognised by a NARROW structural condition on the frames a case sent,
 #: never by an exception type or a case name alone (a mutation can reproduce one by chance). Which of
 #: these a run tolerates is the policy's ``known_defects`` list; tests/test_dast_ingress_sweep.py holds
 #: a strict xfail per entry, so the day a defect is fixed its entry has to come out. Each is matched
 #: against every frame a case sent and explains each frame it matches. None is a handler fault: the
 #: one that was (a blank segment, BACKLOG #1594) is fixed, so a handler-fault NAK is never tolerated.
-KNOWN_DEFECT_DISCRIMINATORS: dict[str, Callable[[bytes], bool]] = {
-    "alphanumeric-field-separator": _alphanumeric_field_separator,
-}
+#: Empty while no defect is open: the alphanumeric field separator was fixed by BACKLOG #2875.
+KNOWN_DEFECT_DISCRIMINATORS: dict[str, Callable[[bytes], bool]] = {}
 #: The only detectors a known defect may silence. A liveness, time, resource or log finding on the
 #: same case is never tolerated: those would be a SECOND defect riding on the known one.
 KNOWN_DEFECT_DETECTORS = frozenset({"reply", "count_and_log"})

@@ -15,6 +15,7 @@
 from __future__ import annotations
 
 import re
+import string
 from collections.abc import Iterator
 from datetime import datetime
 
@@ -39,6 +40,27 @@
 _DEFAULT_FIELD_SEP = "|"
 _DEFAULT_ENC = "^~\\&"
 
+# The characters of the ACK's fixed text: the segment ids, ACK, P, every MSA-1 code (original and
+# enhanced), the letters of the engine's MSA-3 escapes (kept equal to the engine), MSH-7's digits and +/- offset, and the default version 2.5.1. An inbound delimiter from
+# this set would split that text when the ACK is read back, so the ACK is built under the defaults
+# instead (BACKLOG #2875). The engine's messagefoundry/mllpcodec.py holds the same rule, and
+# tests/test_tee_mllp.py pins the two equal.
+_ACK_LITERAL_CHARS = frozenset("MSHMSAACKP" + "EF" + "AAAEARCACECR" + string.digits + "+-.")
+
+
+def _roles(field_sep: str, enc: str) -> dict[str, str]:
+    """Each delimiter character to its role, in MSH-2 order, a short MSH-2 filled from the defaults
+    as the engine's parser fills it. Where two roles share a character the first wins; one with no
+    role is data. The engine's mllpcodec has the same function."""
+    roles: dict[str, str] = {}
+    enc = enc[:4] + _DEFAULT_ENC[len(enc[:4]) :]
+    for char, role in zip(field_sep + enc, "FSRET", strict=True):
+        roles.setdefault(char, role)
+    return roles
+
+
+_DEFAULT_ROLES = _roles(_DEFAULT_FIELD_SEP, _DEFAULT_ENC)
+
 # HL7 segments are CR-delimited, but tolerant peers emit LF/CRLF too — split on any run of either.
 _SEG_SPLIT = re.compile(r"[\r\n]+")
 
@@ -110,25 +132,36 @@ def _msh_fields(message: bytes) -> tuple[str, list[str]]:
     if not first.startswith("MSH") or len(first) < 4:
         return _DEFAULT_FIELD_SEP, []
     field_sep = first[3]
-    return field_sep, first.split(field_sep)
+    # Split after MSH-1, not the whole line: a field separator of M, S or H would split "MSH" too.
+    return field_sep, ["MSH", *first[4:].split(field_sep)]
 
 
 def build_ack(message: bytes, *, code: str = "AA", timestamp: str | None = None) -> bytes:
     """Build an HL7 acknowledgement for ``message`` (default ``AA`` — the relay always accepts).
 
-    Reads the field separator + encoding characters from the inbound MSH (never hardcoded), echoes the
-    control id into MSA-2, swaps sender/receiver so the ACK routes back the way it came, and sanitizes
-    every echoed field against segment injection. If ``message`` isn't parseable HL7 it still returns a
-    generic ``AA`` (the relay always ACKs on receipt). Encoded latin-1 to preserve the inbound's bytes.
+    Reads the field separator + encoding characters from the inbound MSH, echoes the control id into
+    MSA-2, swaps sender/receiver so the ACK routes back the way it came, and sanitizes every echoed
+    field against segment injection. If ``message`` isn't parseable HL7 it still returns a generic
+    ``AA`` (the relay always ACKs on receipt). Encoded latin-1 to preserve the inbound's bytes.
+
+    The one exception to echoing the inbound's delimiters: where its MSH-1 or MSH-2 holds a
+    character of the ACK's fixed text (``_ACK_LITERAL_CHARS``), or two delimiters share a character,
+    the ACK goes out under ``|^~\\&`` so it reads back as an ACK. An echoed value holding either
+    set's escape character, or a default delimiter in a role it does not have in the inbound, is
+    then blanked rather than escaped, by the engine's rule (BACKLOG #2875).
     """
     field_sep, parts = _msh_fields(message)
     enc = _field(parts, 1) or _DEFAULT_ENC
-    sending_app = _no_seg_sep(_field(parts, 2))
-    sending_fac = _no_seg_sep(_field(parts, 3))
-    receiving_app = _no_seg_sep(_field(parts, 4))
-    receiving_fac = _no_seg_sep(_field(parts, 5))
-    control = _no_seg_sep(_field(parts, 9))
-    version = _no_seg_sep(_field(parts, 11)) or "2.5.1"
+    echoed = tuple(_no_seg_sep(_field(parts, index)) for index in (2, 3, 4, 5, 9, 11))
+    roles = _roles(field_sep, enc)
+    # Fewer than five roles means two delimiters share a character, default-filled slots included.
+    if len(roles) < 5 or not _ACK_LITERAL_CHARS.isdisjoint(field_sep + enc):
+        unsafe = {c for c, role in _DEFAULT_ROLES.items() if roles.get(c) != role}
+        unsafe |= {_DEFAULT_ENC[2], enc[2:3] or _DEFAULT_ENC[2]}
+        field_sep, enc = _DEFAULT_FIELD_SEP, _DEFAULT_ENC
+        echoed = tuple(v if unsafe.isdisjoint(v) else "" for v in echoed)
+    sending_app, sending_fac, receiving_app, receiving_fac, control, version = echoed
+    version = version or "2.5.1"
     # MSH-7 carries an explicit numeric UTC offset (BACKLOG #1196), matching the engine's own
     # build_ack. A bare local stamp is ambiguous across a daylight-saving fall-back, where the
     # same wall-clock hour occurs twice, so a partner correlating this acknowledgement against
@@ -174,7 +207,12 @@ def parse_ack(ack: bytes) -> tuple[str | None, str | None]:
             break
     for seg in segments:
         if seg.startswith("MSA"):
-            parts = seg.split(field_sep)
+            # Split after the segment id, not the whole line, as _msh_fields does: a field
+            # separator of M, S or A would split "MSA" too.
+            if seg[3:4] == field_sep:
+                parts = ["MSA", *seg[4:].split(field_sep)]
+            else:
+                parts = seg.split(field_sep)
             return (_field(parts, 1) or None), (_field(parts, 3) or None)
     return None, None
 

@@ -359,6 +359,8 @@ def test_an_unknown_profile_fails_closed() -> None:
 
 def test_a_known_defect_never_silences_liveness_time_or_log() -> None:
     # Judged on the canary path only because it needs no floors; the split happens first either way.
+    # No engine defect is open (the last, BACKLOG #2875, is fixed), so the policy names a stand-in.
+    policy = dict(_policy(), known_defects={"stand-in-defect": "A defect this test pretends to."})
     receipt: dict[str, Any] = {
         "canary": "no-reply",
         "findings": [
@@ -367,12 +369,12 @@ def test_a_known_defect_never_silences_liveness_time_or_log() -> None:
                 "plane": "mllp",
                 "case": "c",
                 "detail": "x",
-                "known_defect": "alphanumeric-field-separator",
+                "known_defect": "stand-in-defect",
             }
             for d in ("reply", "liveness", "time", "log_body")
         ],
     }
-    evaluate(receipt, _policy())
+    evaluate(receipt, policy)
     tolerated = {f["detector"] for f in receipt["known_defect_findings"]}
     assert tolerated == {"reply"}
 
@@ -537,7 +539,8 @@ def test_each_known_defect_discriminator_matches_its_catalogue_case() -> None:
     """A tolerance keyed on a condition no catalogue case carries would tolerate nothing, and one keyed
     on a condition EVERY case carries would tolerate everything."""
     cases = {c.name: c for c in catalogue("ZZTESTSENTINEL", _CAP) if c.plane == "mllp"}
-    anchors = {"alphanumeric-field-separator": "letter-field-separator"}
+    # Empty since BACKLOG #2875 fixed the last open defect; an entry names its anchor case here.
+    anchors: dict[str, str] = {}
     assert set(anchors) == set(KNOWN_DEFECT_DISCRIMINATORS) == set(_policy()["known_defects"])
     for defect, case_name in anchors.items():
         hit = KNOWN_DEFECT_DISCRIMINATORS[defect]
@@ -573,13 +576,9 @@ async def test_a_blank_segment_that_fails_utf8_decode_gets_the_runners_own_nak()
     assert not result.findings, result.findings
 
 
-@pytest.mark.xfail(
-    strict=True,
-    reason=(
-        "ENGINE DEFECT: a message whose MSH-1 is a letter or digit is accepted, and the ACK echoes "
-        "that separator, so MSA-1 (itself letters) cannot be read back by the sender."
-    ),
-)
+# Defect 2, an alphanumeric MSH-1 echoed into an ACK that could not be read back, is FIXED by
+# BACKLOG #2875. Its strict xfail became this plain test and its policy entry came out, so the run
+# no longer tolerates a reply finding on such a frame.
 @_LIVE
 async def test_an_alphanumeric_field_separator_gets_a_readable_reply() -> None:
     result = await _run_alone(_case("mllp", "letter-field-separator"), _budget())