[Bug]: Push directives split across HTTP2 chunks are dropped (alexapy on_message has no reassembly) — PUSH_AUDIO_PLAYER_STATE never parsed, media_player state frozen; patch included

### Describe the bug

`media_player` entities stop following playback: state stays `idle`/`paused` while the Echo is playing, artwork and track never change, and a config entry reload (or `homeassistant.update_entity`) fixes it instantly. Volume, mute, and equalizer changes keep arriving over push the whole time, and `pyscript`/debug inspection shows the HTTP2 stream is alive (ping OK, keepalives flowing). In 13 hours of push uptime the integration parsed **zero** `PUSH_AUDIO_PLAYER_STATE` / `NotifyNowPlayingUpdated` directives while `PUSH_VOLUME_CHANGE` / `PUSH_EQUALIZER_STATE_CHANGE` arrived normally.

Root cause is in `alexapy` (`alexahttp2.py`, present in 1.29.25 and 1.30.0), not the reconnect logic:

- `process_messages()` iterates `response.aiter_text()` and passes **every chunk** to `on_message()`.
- `on_message()` does `for line in message.splitlines(): ... json.loads(line)` inside `contextlib.suppress(JSONDecodeError)`.
- A chunk is an arbitrary slice of the multipart stream, not a whole part. Any directive longer than one chunk arrives as two fragments, each of which fails `json.loads` and is silently discarded. There is no reassembly buffer.

`PUSH_AUDIO_PLAYER_STATE` now carries a `quality` block (badge URL, codec, bitrate, sample rate) on Amazon Music playback, which pushes it past the chunk size. Small directives still fit in one chunk, which is why volume/EQ keep working and why this looks like a "playback state only" freeze. `NotifyNowPlayingUpdated` (~3.5 KB) is affected the same way. This is likely what #3542 is seeing too ("started in the past two weeks", `update_entity` recovers it) and is distinct from the silent-stream case in #3515, which the 1.30.0 read timeout handles.

### To Reproduce

1. Enable debug for `alexapy.alexahttp2` and `custom_components.alexa_media`.
2. Play Amazon Music on an Echo (HD/UHD makes the payload larger).
3. Watch `Received raw message` lines: the `PUSH_AUDIO_PLAYER_STATE` directive is logged in two consecutive chunks, neither of which is followed by a `Received http2push command: PUSH_AUDIO_PLAYER_STATE` line. `media_player` never updates.
4. Call `homeassistant.update_entity` on the player: state and artwork update immediately (polling still works; only push parsing is broken).

### Expected behavior

Directives that span more than one stream chunk are reassembled and delivered to `msg_callback`, so playback state reaches `media_player` over push like volume changes do.

### Screenshots

_No response_

### Home Assistant Version

2026.8.3

### Alexa Media Player Version

5.15.7

### Alexa Media Player API library Version

1.30.0 (also reproduced on 1.29.25)

### Have you entered a 52-character key for Amazon 2SV via Authenticator App login?

No

### Is the HA core integration Alexa Devices installed?

No

### Amazon Domain

amazon.com (US)

### Debug Logs (alexa_media & alexapy)

One directive, two chunks (serials/ids masked). The first chunk ends mid-key (`"sampli`), the second starts with the remainder (`ngRateInHertz"`), and no `Received http2push command` line follows either:

```text
2026-09-02 19:14:29.411 DEBUG (MainThread) [alexapy.alexahttp2] Received raw message: Content-Type: application/json

{"directive":{"header":{"namespace":"Alexa.Mobile.Push","name":"RenderUpdate","messageId":"5********-****-****-****-********a2dc"},"payload":{"renderingUpdates":[{"route":"EventBus:tcomm::message","resourceId":"PUSH_AUDIO_PLAYER_STATE","resourceMetadata":"{\"command\":\"PUSH_AUDIO_PLAYER_STATE\",\"payload\":\"{\\\"dopplerId\\\":{\\\"deviceSerialNumber\\\":\\\"G************4VU\\\",\\\"deviceType\\\":\\\"A3RBAYBE7VM004\\\"},\\\"audioPlayerState\\\":\\\"FINISHED\\\",\\\"quality\\\":{\\\"name\\\":\\\"Ultra High Definition\\\",\\\"badge\\\":{\\\"altText\\\":\\\"https://music-provider-logos.s3.amazonaws.com/badges/AmazonMusic/UHD.png\\\"},\\\"stats\\\":{\\\"codec\\\":\\\"flac\\\",\\\"dataRateInBitsPerSecond\\\":1204442,\\\"sampli
2026-09-02 19:14:29.413 DEBUG (MainThread) [alexapy.alexahttp2] Received raw message: ngRateInHertz\\\":48000}},\\\"error\\\":false,\\\"errorMessage\\\":null,\\\"mediaReferenceId\\\":\\\"a*******-****-****-****-************:3\\\",\\\"destinationUserId\\\":\\\"A**********PAH\\\"}\",\"timeStamp\":1788390869366}"}]}}}
--------abcde123
2026-09-02 19:14:29.628 DEBUG (MainThread) [alexapy.alexahttp2] Received raw message: Content-Type: application/json

{"directive":{ ... "resourceId":"PUSH_VOLUME_CHANGE" ... }
--------abcde123
2026-09-02 19:14:29.629 DEBUG (MainThread) [custom_components.alexa_media] a************y@g*******m: Received http2push command: PUSH_VOLUME_CHANGE : {'dopplerId': {'deviceSerialNumber': 'G************4VU', 'deviceType': 'A3RBAYBE7VM004'}, 'volumeSetting': 47, 'isMuted': False, 'destinationUserId': 'A**********PAH'}
```

Push command history over the 13 hours before the fix (from `hass.data[...]["http2_commands"]`): `PUSH_VOLUME_CHANGE`, `PUSH_EQUALIZER_STATE_CHANGE`, `PUSH_MEDIA_QUEUE_CHANGE`, `NotifyMediaSessionsUpdated`, `PUSH_NOTIFICATION_CHANGE`, `PUSH_DND_STATE_CHANGE` — every one of them small. No `PUSH_AUDIO_PLAYER_STATE`, `PUSH_MEDIA_CHANGE`, or `NotifyNowPlayingUpdated` at all, with music playing.

After patching `alexapy` as below, the very next track change came through as `PUSH_AUDIO_PLAYER_STATE: FINISHED` → `PLAYING` → `NotifyNowPlayingUpdated`, and the entity switched title and artwork with no poll.

### Additional context

**Fix (alexapy `alexahttp2.py`)** — carry the unterminated tail of each chunk into the next and only parse a line once its newline has arrived. Tested locally on 1.30.0 with the chunk-reassembly tests below plus the existing `test_http2_stale_stream.py` (13 passed), `ruff` and `pydocstyle` clean. I do not have a GitLab account for the alexapy MR, so posting the diff here; happy for anyone to carry it over.

```diff
diff --git a/alexapy/alexahttp2.py b/alexapy/alexahttp2.py
index 6b85ab8..ae23fe6 100644
--- a/alexapy/alexahttp2.py
+++ b/alexapy/alexahttp2.py
@@ -108,6 +108,7 @@ class HTTP2EchoClient:
self._last_ping = datetime.datetime(1, 1, 1)
self._last_activity: datetime.datetime | None = None
self._read_timeout = read_timeout
+ self._pending_line: str = ""
self._tasks = set()
self._opened: asyncio.Event = asyncio.Event()
self._closing: bool = False
@@ -233,11 +234,19 @@ class HTTP2EchoClient:
self.on_close("HTTP2 stream ended")

async def on_message(self, message: str) -> None:
- """Handle New Message."""
+ """Handle New Message.
+
+ A chunk is an arbitrary slice of the multipart stream, not a whole
+ part, so a directive longer than one chunk arrives split across
+ several calls. Each line is only parsed once its terminating newline
+ has arrived; the unterminated tail is carried over to the next chunk.
+ """
reauth_required = "Unable to authenticate the request. Please provide a valid authorization token." # noqa: E501
_LOGGER.debug("Received raw message: %s", message)
self._last_activity = datetime.datetime.now(datetime.UTC)
- for line in message.splitlines():
+ *lines, self._pending_line = (self._pending_line + message).split("\n")
+ for raw_line in lines:
+ line = raw_line.rstrip("\r")
if line.startswith("------"):
if not self.boundary: # set boundary character
self.boundary = line
@@ -249,6 +258,12 @@ class HTTP2EchoClient:
elif line and not line.startswith(self.boundary):
with contextlib.suppress(json.decoder.JSONDecodeError):
self._schedule(self.msg_callback(json.loads(line)))
+ # An auth failure body is not part of the multipart stream and may
+ # never be newline-terminated, so don't hold it for the next chunk.
+ if self._pending_line.startswith(reauth_required):
+ _LOGGER.debug("HTTP2 login error: %s", message)
+ self._pending_line = ""
+ await self.handle_login_error("HTTP2 Message Parsing reauth")

def _schedule(self, coro: Coroutine[Any, Any, Any]) -> None:
"""Schedule a coroutine on the target loop safely."""
```

**Tests** (`tests/test_http2_chunked_directives.py`): a 1.5 KB directive fed in 1024-byte chunks is delivered once and intact; small directives interleaved with large ones still parse; a chunk edge inside the boundary line does not register a bogus boundary; CRLF line endings still parse; an unterminated reauth body still triggers `handle_login_error`. All five fail on unpatched 1.30.0.

```python
"""Tests for chunk reassembly in HTTP2EchoClient.on_message.

The failure these tests cover: the directives stream is delivered as
arbitrary chunks, not whole multipart parts. A directive longer than one
chunk (PUSH_AUDIO_PLAYER_STATE grew past that once Amazon Music began
attaching a "quality" block) arrived as two fragments that each failed to
parse as JSON and were silently discarded, so playback state never reached
the consumer while small pushes such as PUSH_VOLUME_CHANGE still did.
"""

import asyncio
import json
from unittest.mock import AsyncMock, MagicMock

import pytest

from alexapy.alexahttp2 import HTTP2EchoClient

BOUNDARY = "--------abcde123"

def _make_client() -> HTTP2EchoClient:
login = MagicMock()
login.session = object()
login.url = "amazon.com"
login.access_token = "test-token"
return HTTP2EchoClient(
login,
msg_callback=AsyncMock(),
open_callback=AsyncMock(),
close_callback=AsyncMock(),
error_callback=AsyncMock(),
loop=asyncio.get_event_loop(),
)

def _part(directive: dict, newline: str = "\n") -> str:
return (
f"{BOUNDARY}{newline}Content-Type: application/json{newline}{newline}"
f"{json.dumps(directive)}{newline}"
)

def _directive(command: str, size: int = 0) -> dict:
return {
"directive": {
"header": {"name": "RenderUpdate"},
"payload": {
"renderingUpdates": [
{"resourceId": command, "resourceMetadata": "x" * size}
]
},
}
}

async def _feed(client: HTTP2EchoClient, stream: str, chunk_size: int) -> list:
for start in range(0, len(stream), chunk_size):
await client.on_message(stream[start : start + chunk_size])
await asyncio.sleep(0) # let scheduled msg_callback tasks run
return [call.args[0] for call in client.msg_callback.await_args_list]

@pytest.mark.asyncio
async def test_directive_split_across_chunks_is_delivered_once():
"""A directive longer than one chunk is reassembled, not dropped."""
client = _make_client()
big = _directive("PUSH_AUDIO_PLAYER_STATE", size=1500)
stream = _part(big) + BOUNDARY + "\n"

received = await _feed(client, stream, chunk_size=1024)

assert received == [big]
assert client._pending_line == ""

@pytest.mark.asyncio
async def test_small_directives_still_parse_alongside_large_ones():
"""Reassembly does not disturb parts that already fit in one chunk."""
client = _make_client()
big = _directive("PUSH_AUDIO_PLAYER_STATE", size=1500)
small = _directive("PUSH_VOLUME_CHANGE")
stream = _part(small) + _part(big) + _part(small) + BOUNDARY + "\n"

received = await _feed(client, stream, chunk_size=1024)

assert received == [small, big, small]

@pytest.mark.asyncio
async def test_boundary_split_across_chunks_is_still_recognized():
"""A chunk edge inside the boundary line must not create a bogus boundary."""
client = _make_client()
small = _directive("PUSH_VOLUME_CHANGE")
stream = _part(small) + BOUNDARY + "\n"
# Chunk edge lands four characters into the first boundary line.
chunks = [stream[:4], stream[4:]]

for chunk in chunks:
await client.on_message(chunk)
await asyncio.sleep(0)

assert client.boundary == BOUNDARY
assert [c.args[0] for c in client.msg_callback.await_args_list] == [small]

@pytest.mark.asyncio
async def test_crlf_lines_are_parsed():
"""CRLF line endings are tolerated as splitlines() tolerated them."""
client = _make_client()
small = _directive("PUSH_VOLUME_CHANGE")
stream = _part(small, newline="\r\n") + BOUNDARY + "\r\n"

received = await _feed(client, stream, chunk_size=7)

assert received == [small]
assert client.boundary == BOUNDARY

@pytest.mark.asyncio
async def test_unterminated_reauth_body_is_not_held_back():
"""An auth failure body without a trailing newline still triggers relogin."""
client = _make_client()
client.handle_login_error = AsyncMock()

await client.on_message(
"Unable to authenticate the request. "
"Please provide a valid authorization token."
)

client.handle_login_error.assert_awaited_once()
assert client._pending_line == ""
```