fix(signal): redistribute textStyle ranges across split message chunks

split_message can break a long Signal payload into multiple JSON-RPC sends,
but the previous code attached the full textStyle list only to chunk 0.
Style ranges in later chunks were dropped, and ranges whose offsets pointed
past chunk 0's end were sent as invalid metadata against chunk 0.

Add _partition_styles, which rebases each range against the chunk it lives
in (in UTF-16 code units, matching the markdown converter) and splits
boundary-spanning ranges across the chunks they touch. Whitespace trimmed
by split_message's lstrip is skipped so offsets stay aligned.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
Kaloyan Tenchov
2026-05-21 01:00:36 +08:00
committed by Xubin Ren
co-authored by Claude Opus 4.7
parent 8f6b7611a2
commit 96eb3b7194
3 changed files with 192 additions and 3 deletions
+32
View File
@@ -834,6 +834,38 @@ class TestSend:
assert "textStyle" in params
assert any("BOLD" in s for s in params["textStyle"])
@pytest.mark.asyncio
async def test_send_split_message_redistributes_text_styles(self):
"""Long message split across chunks: each chunk gets its own textStyle
with offsets rebased to that chunk."""
ch, client = self._make_send_channel()
ch._MAX_MESSAGE_LEN = 12 # type: ignore[attr-defined]
msg = OutboundMessage(
channel="signal",
chat_id="+19995550001",
content="**head** middle and **tail**",
)
await ch.send(msg)
assert len(client.posts) >= 2
# Chunk 0 has BOLD for "head"; chunk 1+ must also carry BOLD for "tail".
bold_chunks = [
p["json"]["params"]
for p in client.posts
if any("BOLD" in s for s in p["json"]["params"].get("textStyle", []))
]
assert len(bold_chunks) >= 2, (
"expected BOLD ranges in more than one chunk; got "
f"{[p['json']['params'] for p in client.posts]}"
)
# Each emitted range must point inside its own chunk's text.
for params in bold_chunks:
chunk_text = params["message"]
for entry in params["textStyle"]:
s, ln, _ = entry.split(":", 2)
start, length = int(s), int(ln)
end_units = start + length
assert end_units <= len(chunk_text.encode("utf-16-le")) // 2
@pytest.mark.asyncio
async def test_send_empty_content_skips_rpc(self):
ch, client = self._make_send_channel()