test(e2e/platforms): port the contract suite onto the PM test environment

- The adapters' SDKs ship in the `messaging` extra, which the default PM
  test environment ([all]) does not carry, so every gateway child timed out
  waiting for its adapter. The rig now fails fast naming the missing
  distribution and the supported fix (`source ./activate --test-extras
  all,messaging`; CI already passes it through setup-pm extras).
- rejected_finalize_leaves_one_copy settles on the ending being visible
  whole OR completed by a continuation message, then asserts first that the
  continuation did not resume mid-word. Main's Slack adapter now reopens a
  sealed stream (message_not_in_streaming_state) with only the text past the
  sealed message, which splits the straddling word: gated on #124219. The
  #95430 partial-copy gate is dropped (fixed on main, issue closed).
- Slack stream_trailing_whitespace is a strict cell again: main's
  edge-whitespace-tolerant finalize (72893ca6a1) makes it pass (#121326).

tests/e2e/core/platforms/_contract.py

@@ -221,10 +221,17 @@ def rejected_finalize_leaves_one_copy(rig: Rig, tag: str, *, group: bool = False
     faults = rig.drv.fail_finalize(lambda text: foot(aid) in norm(text), group=group)
     text, user = f"stream it [in:{token}]", rig.fresh_user()
     inbound = rig.drv.group(text, mention=True) if group else rig.drv.dm(text, user_id=user)
-    wait_reply(rig, inbound.chat_id, aid)
+    # settled once the ending is visible: whole, or completed by a continuation message
+    wait_until(lambda: (any(foot(aid) in norm(v.text) for v in rig.drv.visible(inbound.chat_id))
+                        or _split_seam(rig.drv.visible(inbound.chat_id), aid)),
+               f"reply {aid} visible in {inbound.chat_id}", timeout=TURN_TIMEOUT, on_timeout=rig.ctx)
     barrier(rig, f"finb-{tag}", group=group, user_id=user, attempts=3)
     rig.drv.standin.clear_faults()
     assert any(f.fired for f in faults), f"the finalize fault never fired: nothing was rejected\n{rig.ctx()}"
+    with rig.gate():
+        seam = _split_seam(rig.drv.visible(inbound.chat_id), aid)
+        assert seam is None, (f"the continuation after a rejected finalize resumed mid-word: the answer's "
+                              f"last word is split across two messages ({seam})\n{rig.ctx()}")
     shown = _shown(rig, inbound.chat_id, aid, "q")
     ctx = f"visible: {shown[:300]!r} ... {shown[-300:]!r}\n{rig.ctx()}"
     # the ending is shown exactly once: neither lost nor re-sent by a second final delivery
@@ -238,6 +245,17 @@ def rejected_finalize_leaves_one_copy(rig: Rig, tag: str, *, group: bool = False
             f"(head shown {shown.count(head(aid))}x)\n{ctx}")
 
 
+def _split_seam(visible: List[Visible], aid: str) -> Optional[str]:
+    """The seam, as the user reads it, where the answer's ending is cut mid-word across two
+    consecutive messages (a continuation that resumed mid-word instead of at a word boundary)."""
+    texts, end = [norm(v.text) for v in visible], foot(aid)
+    for a, b in zip(texts, texts[1:]):
+        for i in range(1, len(end)):
+            if a.endswith(end[:i]) and b.startswith(end[i:]):
+                return f"{a[-40:]!r} | {b[:40]!r}"
+    return None
+
+
 def _pwords(text: str, prefix: str) -> List[str]:
     return re.findall(rf"\b{prefix}\d{{3}}\b", text)
 

tests/e2e/core/platforms/_suite.py

@@ -12,7 +12,9 @@ A test module declares ``KNOWN`` (scenario -> (pattern, "#<issue> <symptom>")) a
 
 from __future__ import annotations
 
+import importlib.metadata
 import os
+import sys
 from pathlib import Path
 from typing import Any, Callable, Dict, List, Optional, Tuple
 
@@ -45,6 +47,10 @@ SCENARIOS: Dict[str, Tuple[str, Callable[..., None]]] = {
 }
 
 
+# platform -> the SDK distribution its adapter imports (all from the `messaging` extra)
+_SDK = {"telegram": "python-telegram-bot", "discord": "discord.py", "slack": "slack-bolt"}
+
+
 def scenario_params(skip: Optional[Dict[str, str]] = None) -> List[Any]:
     return [pytest.param(name, id=name, marks=[pytest.mark.skip(reason=skip[name])] if skip and name in skip else [])
             for name in SCENARIOS]
@@ -78,6 +84,15 @@ def rig_fixtures(driver_cls: type) -> Tuple[Any, Any]:
     """``(rig, rig_stream)`` module-scoped fixtures for ``driver_cls``."""
 
     def _make(factory: pytest.TempPathFactory, label: str, extra_cfg: Dict[str, Any], extra_env: Dict[str, str]):
+        # The gateway child runs this interpreter: without the adapter's SDK it only times out later.
+        # (A distribution lookup: the unit-test conftest may leave an SDK stub in sys.modules.)
+        sdk = _SDK[driver_cls.name]
+        try:
+            importlib.metadata.version(sdk)
+        except importlib.metadata.PackageNotFoundError:
+            pytest.fail(f"{sdk} is not installed for {sys.executable}: the test environment lacks the "
+                        "`messaging` extra (`source ./activate --test-extras all,messaging`; CI passes it "
+                        "through setup-pm `extras`)", pytrace=False)
         drv = driver_cls()
         drv.start()
         director = Director()

tests/e2e/core/platforms/test_slack_contract.py

@@ -21,14 +21,12 @@ pytestmark = [
     pytest.mark.skipif(sys.platform == "win32", reason="POSIX process-group gateway harness"),
 ]
 
-_PARTIAL = (r"a partial copy of the answer was left visible after a rejected finalize",
-            "#95430 a rejected closing appendStream/stopStream leaves the partial stream next to the re-posted answer")
+_MIDWORD = (r"the continuation after a rejected finalize resumed mid-word",
+            "#124219 a stream reopened after message_not_in_streaming_state resumes mid-word (word split across "
+            "two messages)")
 KNOWN: dict[str, tuple[str, str]] = {
-    "stream_finalize_rejected": _PARTIAL,
-    "stream_finalize_rejected_group": _PARTIAL,
-    "stream_trailing_whitespace": (
-        r"a streamed reply ending in whitespace is shown != once",
-        "#121326 native streaming re-posts the whole reply when it ends in whitespace"),
+    "stream_finalize_rejected": _MIDWORD,
+    "stream_finalize_rejected_group": _MIDWORD,
     "planned_restart_notice": (
         r"a redelivered /restart restarted the gateway again|a second restart ack means the replayed /restart was obeyed",
         "#121325 a replayed /restart restarts the gateway again (guard needs Telegram update ids)"),