From d2a2a83dcc10dc5d08ec754a91245524078bc73a Mon Sep 17 00:00:00 2001
From: TimeToBuildBob <timetobuildbob@gmail.com>
Date: Fri, 2 Oct 2026 04:40:08 +0000
Subject: [PATCH 1/9] feat(queue): merge queued heartbeats per bucket before
 dispatch

When aw-server is unreachable the request queue accumulates one heartbeat per
commit interval. On reconnect they were dispatched one HTTP request at a time,
so a long outage drained slowly and hammered the server just as it came back
(issues #32 and #7).

RequestQueue now pops queued requests in batches and pre-merges consecutive
heartbeats for the same bucket client-side with aw_transform.heartbeat_merge
before sending. The merged events still go to the heartbeat endpoint, so the
server-side result is unchanged - a run of identical heartbeats that the
server would have merged into one event now arrives as one request instead of
dozens.

Requests are held in an in-memory batch until the whole batch has been
handled; a transient error retains and retries the batch, and re-sending an
already-delivered heartbeat is a no-op on the server. A non-retryable client
error drops only that request and keeps dispatching the rest.

Test fills the queue with 200 mergeable heartbeats and asserts the drain uses
a single request while the merged event still spans the whole range (no data
loss).

Git-Session-Id: b337
---
 aw_client/client.py        | 203 +++++++++++++++++++++++++++----------
 tests/test_requestqueue.py | 148 +++++++++++++++++++++++++++
 2 files changed, 296 insertions(+), 55 deletions(-)

diff --git a/aw_client/client.py b/aw_client/client.py
index 1cacba4..f8efc0d 100644
--- a/aw_client/client.py
+++ b/aw_client/client.py
@@ -4,6 +4,7 @@
 import json
 import logging
 import os
+import re
 import socket
 import sqlite3
 import threading
@@ -593,6 +594,23 @@ def _retry_delay(response: req.Response) -> float:
     return max(RETRY_DELAY_DEFAULT, min(delay, RETRY_DELAY_MAX))
 
 
+# Matches the endpoint the queue stores heartbeats under, capturing the bucket
+# id and the pulsetime needed to merge consecutive queued heartbeats.
+_HEARTBEAT_ENDPOINT_RE = re.compile(
+    r"^buckets/(?P<bucket_id>.+)/heartbeat\?pulsetime=(?P<pulsetime>[0-9.]+)$"
+)
+
+
+def _try_event(data: Any) -> Optional[Event]:
+    """Parse a queued payload as an Event, or None if it is not one."""
+    if not isinstance(data, dict):
+        return None
+    try:
+        return Event(**data)
+    except TypeError:
+        return None
+
+
 class RequestQueue(threading.Thread):
     """Used to asynchronously send heartbeats.
 
@@ -608,6 +626,10 @@ class RequestQueue(threading.Thread):
     # 503 in particular is sent by aw-server when the heartbeat lock times out.
     RETRY_STATUS_CODES = {429, 500, 502, 503, 504}
 
+    # How many queued requests to pop at once, so consecutive heartbeats for
+    # the same bucket can be merged before dispatch.
+    BATCH_SIZE = 1000
+
     def __init__(
         self,
         client: ActivityWatchClient,
@@ -648,21 +670,41 @@ def __init__(
         self._persistqueue = persistqueue.FIFOSQLiteQueue(
             persistqueue_path, multithreading=True, auto_commit=False
         )
-        self._current = None  # type: Optional[QueuedRequest]
+        # The requests popped from the queue but not yet acknowledged. Held
+        # across dispatches so a transient error retries them instead of
+        # dropping them. A single task_done() deletes every row <= the cursor,
+        # i.e. the whole batch, so no per-item bookkeeping is needed.
+        self._current_batch = []  # type: List[QueuedRequest]
         self._queue_write_failing = False
 
     def _get_next(self) -> Optional[QueuedRequest]:
-        # self._current will always hold the next not-yet-sent event,
-        # until self._task_done() is called.
-        if not self._current:
+        # Returns the head of the in-flight batch, otherwise pops a single
+        # request. Used by tests/callers that only need one item.
+        if not self._current_batch:
             try:
-                self._current = self._persistqueue.get(block=False)
+                self._current_batch = [self._persistqueue.get(block=False)]
             except persistqueue.exceptions.Empty:
                 return None
-        return self._current
+        return self._current_batch[0]
+
+    def _get_next_batch(self) -> List[QueuedRequest]:
+        # Pop up to BATCH_SIZE requests in one go so consecutive heartbeats
+        # for the same bucket can be merged before dispatch. The in-memory
+        # batch is only acknowledged (deleted) by _task_done(), so a crash
+        # mid-batch leaves every popped request on disk for the next run.
+        if self._current_batch:
+            return self._current_batch
+        batch = []  # type: List[QueuedRequest]
+        while len(batch) < self.BATCH_SIZE:
+            try:
+                batch.append(self._persistqueue.get(block=False))
+            except persistqueue.exceptions.Empty:
+                break
+        self._current_batch = batch
+        return batch
 
     def _task_done(self) -> None:
-        self._current = None
+        self._current_batch = []
         self._persistqueue.task_done()
 
     def _create_buckets(self) -> None:
@@ -699,10 +741,10 @@ def wait_for_queue_empty(self, timeout: Optional[float] = None) -> bool:
         :return: True if the queue became empty, False if the timeout was reached.
         """
         if not self.is_alive():
-            return self._persistqueue.qsize() == 0 and self._current is None
+            return self._persistqueue.qsize() == 0 and not self._current_batch
 
         start_time = monotonic()
-        while self._persistqueue.qsize() > 0 or self._current is not None:
+        while self._persistqueue.qsize() > 0 or self._current_batch:
             if timeout is not None and monotonic() - start_time >= timeout:
                 return False
             if self.wait(0.1):
@@ -710,60 +752,111 @@ def wait_for_queue_empty(self, timeout: Optional[float] = None) -> bool:
                 return False
         return True
 
+    def _coalesce(self, batch: List[QueuedRequest]) -> List[QueuedRequest]:
+        """Merge consecutive queued heartbeats for the same bucket.
+
+        When the server is unreachable the queue accumulates one heartbeat per
+        commit interval. Consecutive heartbeats for a bucket usually carry
+        identical data, so aw_transform.heartbeat_merge collapses them into a
+        handful of long events. Sending the merged events to the heartbeat
+        endpoint produces the same server-side result with far fewer requests.
+        """
+        groups = {}  # type: Dict[str, Tuple[Optional[float], List[Dict[str, Any]]]]
+        for request in batch:
+            if request.endpoint not in groups:
+                match = _HEARTBEAT_ENDPOINT_RE.match(request.endpoint)
+                pulsetime = float(match.group("pulsetime")) if match else None
+                groups[request.endpoint] = (pulsetime, [])
+            groups[request.endpoint][1].append(request.data)
+
+        coalesced = []  # type: List[QueuedRequest]
+        for endpoint, (pulsetime, datas) in groups.items():
+            if pulsetime is None:
+                # Unknown endpoint shape: send verbatim, unmerged.
+                coalesced.extend(QueuedRequest(endpoint, data) for data in datas)
+                continue
+            merged = []  # type: List[Event]
+            for data in datas:
+                event = _try_event(data)
+                if event is None:
+                    # Non-event payload: flush what has been merged and send
+                    # this one verbatim.
+                    coalesced.extend(
+                        QueuedRequest(endpoint, e.to_json_dict()) for e in merged
+                    )
+                    merged = []
+                    coalesced.append(QueuedRequest(endpoint, data))
+                    continue
+                if merged:
+                    merged_event = heartbeat_merge(merged[-1], event, pulsetime)
+                    if merged_event is not None:
+                        merged[-1] = merged_event
+                        continue
+                merged.append(event)
+            coalesced.extend(
+                QueuedRequest(endpoint, event.to_json_dict()) for event in merged
+            )
+        return coalesced
+
     def _dispatch_request(self) -> None:
-        request = self._get_next()
-        if not request:
+        batch = self._get_next_batch()
+        if not batch:
             self.wait(0.2)  # seconds to wait before re-polling the empty queue
             return
 
-        try:
-            self.client._post(request.endpoint, request.data)
-        except (req.exceptions.ConnectionError, req.exceptions.Timeout):
-            # Triggered by:
-            #   - server not running (connection refused)
-            #   - server not responding (timeout)
-            # Keep the request in the queue and go back to waiting for the
-            # server to become available (the run loop reconnects).
-            self.connected = False
-            logger.warning(
-                "Connection refused or timeout, will queue requests until connection is available."
-            )
-            # wait a bit before retrying, so we don't spam the server (or logs), see:
-            #  - https://github.com/ActivityWatch/activitywatch/issues/815
-            #  - https://github.com/ActivityWatch/activitywatch/issues/756#issuecomment-1266662861
-            sleep(0.5)
-            return
-        except req.RequestException as e:
-            # NOTE: `e.response is not None` matters: Response.__bool__ is
-            # False for any non-2xx status, so a plain `if e.response` never
-            # matches an error response.
-            response = e.response
-            status_code = response.status_code if response is not None else None
-            if response is not None and status_code in self.RETRY_STATUS_CODES:
-                # Transient server-side problem (busy, overloaded, restarting
-                # or behind a flaky proxy) - the request itself is likely
-                # fine, so keep it in the queue and retry. Heartbeats are safe
-                # to replay: a duplicate of an already-processed heartbeat
-                # merges into the last event as a no-op.
-                delay = _retry_delay(response)
+        for request in self._coalesce(batch):
+            try:
+                self.client._post(request.endpoint, request.data)
+            except (req.exceptions.ConnectionError, req.exceptions.Timeout):
+                # Triggered by:
+                #   - server not running (connection refused)
+                #   - server not responding (timeout)
+                # Keep the whole batch in memory and go back to waiting for the
+                # server to become available (the run loop reconnects).
+                # Re-sending already-delivered heartbeats is safe: the server
+                # merges a duplicate into the last event as a no-op.
+                self.connected = False
                 logger.warning(
-                    f"Server error {status_code}, will retry in {delay}s: {request.endpoint}"
+                    "Connection refused or timeout, will queue requests until connection is available."
                 )
-                # stop-aware wait, so a long Retry-After can't block shutdown
-                self.wait(delay)
+                # wait a bit before retrying, so we don't spam the server (or logs), see:
+                #  - https://github.com/ActivityWatch/activitywatch/issues/815
+                #  - https://github.com/ActivityWatch/activitywatch/issues/756#issuecomment-1266662861
+                sleep(0.5)
                 return
-            else:
-                # Client errors (e.g. HTTP 400 - bad request, see
-                # https://github.com/ActivityWatch/activitywatch/issues/815)
-                # are likely to fail forever, so drop the request instead of
-                # blocking the queue.
-                logger.exception(
-                    f"Request failed ({status_code}), not retrying: {request.data}"
-                )
-        except Exception:
-            logger.exception(f"Unknown error, not retrying: {request.data}")
+            except req.RequestException as e:
+                # NOTE: `e.response is not None` matters: Response.__bool__ is
+                # False for any non-2xx status, so a plain `if e.response` never
+                # matches an error response.
+                response = e.response
+                status_code = response.status_code if response is not None else None
+                if response is not None and status_code in self.RETRY_STATUS_CODES:
+                    # Transient server-side problem (busy, overloaded,
+                    # restarting or behind a flaky proxy) - the request itself
+                    # is likely fine, so keep the batch and retry. Heartbeats
+                    # are safe to replay: a duplicate of an already-processed
+                    # heartbeat merges into the last event as a no-op.
+                    delay = _retry_delay(response)
+                    logger.warning(
+                        f"Server error {status_code}, will retry in {delay}s: {request.endpoint}"
+                    )
+                    # stop-aware wait, so a long Retry-After can't block shutdown
+                    self.wait(delay)
+                    return
+                else:
+                    # Client errors (e.g. HTTP 400 - bad request, see
+                    # https://github.com/ActivityWatch/activitywatch/issues/815)
+                    # are likely to fail forever, so drop this request and
+                    # keep dispatching the rest of the batch. The remaining
+                    # requests are not lost: the batch is only acknowledged
+                    # once every request has been handled.
+                    logger.error(
+                        f"Request failed ({status_code}), not retrying: {request.data}"
+                    )
+            except Exception:
+                logger.exception(f"Unknown error, not retrying: {request.data}")
 
-        # Mark the request as done
+        # Mark the whole batch as done
         self._task_done()
 
     def run(self) -> None:
diff --git a/tests/test_requestqueue.py b/tests/test_requestqueue.py
index 67afcb6..0797141 100644
--- a/tests/test_requestqueue.py
+++ b/tests/test_requestqueue.py
@@ -8,6 +8,7 @@
 """
 
 import threading
+from datetime import datetime, timedelta, timezone
 from time import sleep
 from logging import basicConfig, DEBUG, WARNING
 
@@ -18,6 +19,16 @@
 
 from aw_client.client import RequestQueue
 
+_BASE = datetime(2026, 1, 1, tzinfo=timezone.utc)
+
+
+def _heartbeat(offset_s, duration_s=10.0, data=None):
+    return {
+        "timestamp": (_BASE + timedelta(seconds=offset_s)).isoformat(),
+        "duration": duration_s,
+        "data": {"status": "not-afk"} if data is None else data,
+    }
+
 
 class MockClient:
     client_name = "Mock"
@@ -290,3 +301,140 @@ def resp(retry_after=None):
     assert _retry_delay(resp("0")) == RETRY_DELAY_DEFAULT  # floored
     # HTTP-date form is not parsed, falls back to default
     assert _retry_delay(resp("Wed, 21 Oct 2026 07:28:00 GMT")) == RETRY_DELAY_DEFAULT
+
+
+class RecordingClient(MockClient):
+    """Client that records every (endpoint, data) posted to it."""
+
+    def __init__(self):
+        super().__init__()
+        self.posts = []
+        self.post_calls = 0
+
+    def _post(self, endpoint, data, **kwargs):
+        self.posts.append((endpoint, data))
+        self.post_calls += 1
+        return requests.Response()
+
+
+def test_dispatch_merges_consecutive_queued_heartbeats():
+    """
+    A long offline backlog of heartbeats for one bucket must collapse into a
+    handful of requests instead of one per heartbeat, without losing data
+    (issue #32 / #7).
+    """
+    client = RecordingClient()
+    rq = _fresh_queue(client)
+    rq.connected = True
+
+    n = 200
+    for i in range(n):
+        rq.add_request("buckets/test/heartbeat?pulsetime=10", _heartbeat(i * 10))
+
+    rq._dispatch_request()
+
+    assert client.post_calls == 1, (
+        "mergeable heartbeats should collapse into one request"
+    )
+    assert rq._get_next() is None, "the queue should be fully drained"
+    endpoint, data = client.posts[0]
+    assert endpoint == "buckets/test/heartbeat?pulsetime=10"
+    # No data loss: the merged event spans the whole queued range.
+    assert data["duration"] == pytest.approx(n * 10)
+    assert data["data"] == {"status": "not-afk"}
+
+
+def test_dispatch_does_not_merge_heartbeats_with_different_data():
+    """Heartbeats whose data differs must each be sent, in order."""
+    client = RecordingClient()
+    rq = _fresh_queue(client)
+    rq.connected = True
+
+    for i in range(5):
+        rq.add_request(
+            "buckets/test/heartbeat?pulsetime=10",
+            _heartbeat(i * 10, data={"title": f"window-{i}"}),
+        )
+
+    rq._dispatch_request()
+
+    assert client.post_calls == 5
+    assert [d["data"]["title"] for _, d in client.posts] == [
+        f"window-{i}" for i in range(5)
+    ]
+    assert rq._get_next() is None
+
+
+def test_dispatch_batches_across_buckets():
+    """Interleaved buckets are grouped, so each bucket drains in one request."""
+    client = RecordingClient()
+    rq = _fresh_queue(client)
+    rq.connected = True
+
+    for i in range(50):
+        rq.add_request("buckets/afk/heartbeat?pulsetime=10", _heartbeat(i * 10))
+        rq.add_request(
+            "buckets/window/heartbeat?pulsetime=10",
+            _heartbeat(i * 10, data={"title": "same"}),
+        )
+
+    rq._dispatch_request()
+
+    assert client.post_calls == 2
+    assert sorted(endpoint for endpoint, _ in client.posts) == [
+        "buckets/afk/heartbeat?pulsetime=10",
+        "buckets/window/heartbeat?pulsetime=10",
+    ]
+    assert rq._get_next() is None
+
+
+def test_dispatch_retry_keeps_whole_batch_then_merges():
+    """A transient error retains the whole batch; the retry merges it."""
+    client = FlakyClient(_http_error(503))
+    rq = _fresh_queue(client)
+    rq.connected = True
+
+    n = 50
+    for i in range(n):
+        rq.add_request("buckets/test/heartbeat?pulsetime=10", _heartbeat(i * 10))
+
+    rq._dispatch_request()
+
+    assert client.post_calls == 1
+    assert rq._get_next() is not None  # batch retained
+
+    client.exc = None  # server recovered
+    rq._dispatch_request()
+
+    assert client.post_calls == 2  # one merged request on retry
+    assert rq._get_next() is None
+
+
+def test_dispatch_drops_bad_request_but_keeps_the_rest():
+    """
+    A non-retryable error (HTTP 400) drops only that request; the rest of the
+    batch must still be dispatched and the queue fully drained.
+    """
+
+    class SelectiveFailClient(RecordingClient):
+        def _post(self, endpoint, data, **kwargs):
+            self.post_calls += 1
+            if "bad" in endpoint:
+                raise _http_error(400)
+            self.posts.append((endpoint, data))
+            return requests.Response()
+
+    client = SelectiveFailClient()
+    rq = _fresh_queue(client)
+    rq.connected = True
+
+    rq.add_request("buckets/bad/heartbeat?pulsetime=10", _heartbeat(0))
+    rq.add_request("buckets/good/heartbeat?pulsetime=10", _heartbeat(0, data={"k": 1}))
+
+    rq._dispatch_request()
+
+    assert client.post_calls == 2
+    assert [endpoint for endpoint, _ in client.posts] == [
+        "buckets/good/heartbeat?pulsetime=10"
+    ]
+    assert rq._get_next() is None

From b6905071b6acf202c0bc8138bf0c54ab3282460c Mon Sep 17 00:00:00 2001
From: TimeToBuildBob <timetobuildbob@gmail.com>
Date: Fri, 2 Oct 2026 04:41:41 +0000
Subject: [PATCH 2/9] test(queue): drain a 10k-heartbeat backlog in batches

Git-Session-Id: b337
---
 tests/test_requestqueue.py | 27 +++++++++++++++++++++++++++
 1 file changed, 27 insertions(+)

diff --git a/tests/test_requestqueue.py b/tests/test_requestqueue.py
index 0797141..365324c 100644
--- a/tests/test_requestqueue.py
+++ b/tests/test_requestqueue.py
@@ -8,6 +8,7 @@
 """
 
 import threading
+import time
 from datetime import datetime, timedelta, timezone
 from time import sleep
 from logging import basicConfig, DEBUG, WARNING
@@ -438,3 +439,29 @@ def _post(self, endpoint, data, **kwargs):
         "buckets/good/heartbeat?pulsetime=10"
     ]
     assert rq._get_next() is None
+
+
+def test_dispatch_drains_10k_heartbeats_in_batches():
+    """
+    A large offline backlog (10k mergeable heartbeats) must drain with one
+    request per batch, quickly and without losing data.
+    """
+    client = RecordingClient()
+    rq = _fresh_queue(client)
+    rq.connected = True
+
+    n = 10_000
+    for i in range(n):
+        rq.add_request("buckets/test/heartbeat?pulsetime=10", _heartbeat(i * 10))
+
+    start = time.monotonic()
+    while rq._get_next_batch():
+        rq._dispatch_request()
+    elapsed = time.monotonic() - start
+
+    # Every batch merges to a single heartbeat request.
+    assert client.post_calls == n // rq.BATCH_SIZE
+    assert rq._get_next() is None
+    # The first request already spans the whole backlog (no data loss).
+    assert client.posts[0][1]["duration"] > 0
+    assert elapsed < 10, f"draining 10k heartbeats took {elapsed:.1f}s"

From fb52b50d6f660567305b3e00833c5e474ac5d790 Mon Sep 17 00:00:00 2001
From: TimeToBuildBob <timetobuildbob@gmail.com>
Date: Fri, 2 Oct 2026 04:46:35 +0000
Subject: [PATCH 3/9] fix(queue): address review: preserve order, guard
 malformed endpoints, per-request retry

Greptile review feedback on #127:

- **Order**: group by bucket, not by full endpoint, and only merge a
  contiguous run that shares one endpoint. Grouping by endpoint reordered a
  bucket's heartbeats when their pulsetimes differed (e.g. 10, 20, 10 was sent
  as 10, 10, 20), which changes the server-side timeline.
- **Malformed endpoint**: `pulsetime=1..2` matched the regex but `float()`
  raised inside coalescing, outside the dispatch error handler, killing the
  worker. Parsing is now fallible and a malformed pulsetime falls back to
  sending the request verbatim.
- **Partial retry**: dispatch one coalesced request per call and track the
  index, so a transient error retries only the failed request instead of
  replaying heartbeats that already reached the server.
- **Shutdown**: because one request is dispatched per call, `disconnect()`
  observes `stop()` between requests instead of blocking on a whole batch.

Tests: 21 passed (new: pulsetime-order preservation, malformed pulsetime,
partial-retry-does-not-replay).

Git-Session-Id: b337
---
 aw_client/client.py        | 198 +++++++++++++++++++++++--------------
 tests/test_requestqueue.py |  92 ++++++++++++++++-
 2 files changed, 211 insertions(+), 79 deletions(-)

diff --git a/aw_client/client.py b/aw_client/client.py
index f8efc0d..39f89e7 100644
--- a/aw_client/client.py
+++ b/aw_client/client.py
@@ -597,10 +597,26 @@ def _retry_delay(response: req.Response) -> float:
 # Matches the endpoint the queue stores heartbeats under, capturing the bucket
 # id and the pulsetime needed to merge consecutive queued heartbeats.
 _HEARTBEAT_ENDPOINT_RE = re.compile(
-    r"^buckets/(?P<bucket_id>.+)/heartbeat\?pulsetime=(?P<pulsetime>[0-9.]+)$"
+    r"^buckets/(?P<bucket_id>.+)/heartbeat\?pulsetime=(?P<pulsetime>[0-9]+(?:\.[0-9]+)?)$"
 )
 
 
+def _parse_heartbeat_endpoint(endpoint: str) -> Optional[Tuple[str, float]]:
+    """Return (bucket_id, pulsetime) for a queued heartbeat endpoint, else None.
+
+    A malformed pulsetime must not raise: it just falls back to sending the
+    request verbatim, so one poisoned endpoint cannot wedge the queue.
+    """
+    match = _HEARTBEAT_ENDPOINT_RE.match(endpoint)
+    if match is None:
+        return None
+    try:
+        pulsetime = float(match.group("pulsetime"))
+    except ValueError:
+        return None
+    return match.group("bucket_id"), pulsetime
+
+
 def _try_event(data: Any) -> Optional[Event]:
     """Parse a queued payload as an Event, or None if it is not one."""
     if not isinstance(data, dict):
@@ -676,6 +692,12 @@ def __init__(
         # i.e. the whole batch, so no per-item bookkeeping is needed.
         self._current_batch = []  # type: List[QueuedRequest]
         self._queue_write_failing = False
+        # `_current_batch` after coalescing, plus how many of those have been
+        # handled. Dispatching one coalesced request per call means a transient
+        # error retries only the failed request - never replays ones that
+        # already reached the server - and lets the run loop observe stop().
+        self._coalesced_batch = []  # type: List[QueuedRequest]
+        self._coalesced_index = 0
 
     def _get_next(self) -> Optional[QueuedRequest]:
         # Returns the head of the in-flight batch, otherwise pops a single
@@ -705,6 +727,8 @@ def _get_next_batch(self) -> List[QueuedRequest]:
 
     def _task_done(self) -> None:
         self._current_batch = []
+        self._coalesced_batch = []
+        self._coalesced_index = 0
         self._persistqueue.task_done()
 
     def _create_buckets(self) -> None:
@@ -760,42 +784,55 @@ def _coalesce(self, batch: List[QueuedRequest]) -> List[QueuedRequest]:
         identical data, so aw_transform.heartbeat_merge collapses them into a
         handful of long events. Sending the merged events to the heartbeat
         endpoint produces the same server-side result with far fewer requests.
+
+        Requests are grouped by bucket, not by full endpoint: grouping by
+        endpoint would reorder a bucket's heartbeats when their pulsetimes
+        differ. Within a bucket only a contiguous run sharing one endpoint is
+        merged, so the order of a bucket's requests is preserved exactly.
         """
-        groups = {}  # type: Dict[str, Tuple[Optional[float], List[Dict[str, Any]]]]
+        groups = {}  # type: Dict[str, List[QueuedRequest]]
         for request in batch:
-            if request.endpoint not in groups:
-                match = _HEARTBEAT_ENDPOINT_RE.match(request.endpoint)
-                pulsetime = float(match.group("pulsetime")) if match else None
-                groups[request.endpoint] = (pulsetime, [])
-            groups[request.endpoint][1].append(request.data)
+            parsed = _parse_heartbeat_endpoint(request.endpoint)
+            key = parsed[0] if parsed is not None else request.endpoint
+            groups.setdefault(key, []).append(request)
 
         coalesced = []  # type: List[QueuedRequest]
-        for endpoint, (pulsetime, datas) in groups.items():
-            if pulsetime is None:
-                # Unknown endpoint shape: send verbatim, unmerged.
-                coalesced.extend(QueuedRequest(endpoint, data) for data in datas)
-                continue
+
+        def flush(endpoint: Optional[str], merged: List[Event]) -> None:
+            coalesced.extend(
+                QueuedRequest(endpoint, event.to_json_dict()) for event in merged
+            )
+
+        for requests in groups.values():
             merged = []  # type: List[Event]
-            for data in datas:
-                event = _try_event(data)
-                if event is None:
-                    # Non-event payload: flush what has been merged and send
-                    # this one verbatim.
-                    coalesced.extend(
-                        QueuedRequest(endpoint, e.to_json_dict()) for e in merged
-                    )
+            merged_endpoint = None  # type: Optional[str]
+            for request in requests:
+                parsed = _parse_heartbeat_endpoint(request.endpoint)
+                event = _try_event(request.data) if parsed is not None else None
+                if (
+                    parsed is None
+                    or event is None
+                    or request.endpoint != merged_endpoint
+                ):
+                    # Unknown endpoint shape, non-event payload, or a new
+                    # endpoint: close the current run and start a new one.
+                    flush(merged_endpoint, merged)
                     merged = []
-                    coalesced.append(QueuedRequest(endpoint, data))
+                    merged_endpoint = None
+                if parsed is None or event is None:
+                    coalesced.append(request)
                     continue
                 if merged:
-                    merged_event = heartbeat_merge(merged[-1], event, pulsetime)
+                    merged_event = heartbeat_merge(merged[-1], event, parsed[1])
                     if merged_event is not None:
                         merged[-1] = merged_event
                         continue
+                    # Not mergeable: close the run before starting a new one.
+                    flush(merged_endpoint, merged)
+                    merged = []
                 merged.append(event)
-            coalesced.extend(
-                QueuedRequest(endpoint, event.to_json_dict()) for event in merged
-            )
+                merged_endpoint = request.endpoint
+            flush(merged_endpoint, merged)
         return coalesced
 
     def _dispatch_request(self) -> None:
@@ -804,60 +841,69 @@ def _dispatch_request(self) -> None:
             self.wait(0.2)  # seconds to wait before re-polling the empty queue
             return
 
-        for request in self._coalesce(batch):
-            try:
-                self.client._post(request.endpoint, request.data)
-            except (req.exceptions.ConnectionError, req.exceptions.Timeout):
-                # Triggered by:
-                #   - server not running (connection refused)
-                #   - server not responding (timeout)
-                # Keep the whole batch in memory and go back to waiting for the
-                # server to become available (the run loop reconnects).
-                # Re-sending already-delivered heartbeats is safe: the server
-                # merges a duplicate into the last event as a no-op.
-                self.connected = False
+        if not self._coalesced_batch:
+            self._coalesced_batch = self._coalesce(batch)
+            if not self._coalesced_batch:
+                self._task_done()
+                return
+
+        # Dispatch one coalesced request per call: a transient error then
+        # retries only the failed request (never replays ones that already
+        # reached the server), and the run loop can observe stop() between
+        # requests instead of blocking on a long batch.
+        request = self._coalesced_batch[self._coalesced_index]
+        try:
+            self.client._post(request.endpoint, request.data)
+        except (req.exceptions.ConnectionError, req.exceptions.Timeout):
+            # Triggered by:
+            #   - server not running (connection refused)
+            #   - server not responding (timeout)
+            # Keep the batch in memory and go back to waiting for the server
+            # to become available (the run loop reconnects).
+            self.connected = False
+            logger.warning(
+                "Connection refused or timeout, will queue requests until connection is available."
+            )
+            # wait a bit before retrying, so we don't spam the server (or logs), see:
+            #  - https://github.com/ActivityWatch/activitywatch/issues/815
+            #  - https://github.com/ActivityWatch/activitywatch/issues/756#issuecomment-1266662861
+            sleep(0.5)
+            return
+        except req.RequestException as e:
+            # NOTE: `e.response is not None` matters: Response.__bool__ is
+            # False for any non-2xx status, so a plain `if e.response` never
+            # matches an error response.
+            response = e.response
+            status_code = response.status_code if response is not None else None
+            if response is not None and status_code in self.RETRY_STATUS_CODES:
+                # Transient server-side problem (busy, overloaded, restarting
+                # or behind a flaky proxy) - the request itself is likely
+                # fine, so retry it. Heartbeats are safe to replay: a
+                # duplicate of an already-processed heartbeat merges into the
+                # last event as a no-op.
+                delay = _retry_delay(response)
                 logger.warning(
-                    "Connection refused or timeout, will queue requests until connection is available."
+                    f"Server error {status_code}, will retry in {delay}s: {request.endpoint}"
                 )
-                # wait a bit before retrying, so we don't spam the server (or logs), see:
-                #  - https://github.com/ActivityWatch/activitywatch/issues/815
-                #  - https://github.com/ActivityWatch/activitywatch/issues/756#issuecomment-1266662861
-                sleep(0.5)
+                # stop-aware wait, so a long Retry-After can't block shutdown
+                self.wait(delay)
                 return
-            except req.RequestException as e:
-                # NOTE: `e.response is not None` matters: Response.__bool__ is
-                # False for any non-2xx status, so a plain `if e.response` never
-                # matches an error response.
-                response = e.response
-                status_code = response.status_code if response is not None else None
-                if response is not None and status_code in self.RETRY_STATUS_CODES:
-                    # Transient server-side problem (busy, overloaded,
-                    # restarting or behind a flaky proxy) - the request itself
-                    # is likely fine, so keep the batch and retry. Heartbeats
-                    # are safe to replay: a duplicate of an already-processed
-                    # heartbeat merges into the last event as a no-op.
-                    delay = _retry_delay(response)
-                    logger.warning(
-                        f"Server error {status_code}, will retry in {delay}s: {request.endpoint}"
-                    )
-                    # stop-aware wait, so a long Retry-After can't block shutdown
-                    self.wait(delay)
-                    return
-                else:
-                    # Client errors (e.g. HTTP 400 - bad request, see
-                    # https://github.com/ActivityWatch/activitywatch/issues/815)
-                    # are likely to fail forever, so drop this request and
-                    # keep dispatching the rest of the batch. The remaining
-                    # requests are not lost: the batch is only acknowledged
-                    # once every request has been handled.
-                    logger.error(
-                        f"Request failed ({status_code}), not retrying: {request.data}"
-                    )
-            except Exception:
-                logger.exception(f"Unknown error, not retrying: {request.data}")
-
-        # Mark the whole batch as done
-        self._task_done()
+            else:
+                # Client errors (e.g. HTTP 400 - bad request, see
+                # https://github.com/ActivityWatch/activitywatch/issues/815)
+                # are likely to fail forever, so drop this request and move on
+                # to the next one.
+                logger.error(
+                    f"Request failed ({status_code}), not retrying: {request.data}"
+                )
+        except Exception:
+            logger.exception(f"Unknown error, not retrying: {request.data}")
+
+        # Handled (delivered or dropped); advance. Acknowledge the batch only
+        # once every request in it has been handled.
+        self._coalesced_index += 1
+        if self._coalesced_index >= len(self._coalesced_batch):
+            self._task_done()
 
     def run(self) -> None:
         self._stop_event.clear()
diff --git a/tests/test_requestqueue.py b/tests/test_requestqueue.py
index 365324c..eb525de 100644
--- a/tests/test_requestqueue.py
+++ b/tests/test_requestqueue.py
@@ -228,6 +228,12 @@ def _fresh_queue(client, tmp_path) -> RequestQueue:
     return RequestQueue(client, persistqueue_path=str(tmp_path / "queue"))  # type: ignore
 
 
+def _drain(rq) -> None:
+    """Dispatch until the queue is empty (only for clients that succeed)."""
+    while rq._get_next_batch():
+        rq._dispatch_request()
+
+
 @pytest.mark.parametrize("status_code", [429, 500, 502, 503, 504])
 def test_dispatch_retries_transient_server_errors(status_code, tmp_path):
     """
@@ -357,7 +363,7 @@ def test_dispatch_does_not_merge_heartbeats_with_different_data():
             _heartbeat(i * 10, data={"title": f"window-{i}"}),
         )
 
-    rq._dispatch_request()
+    _drain(rq)
 
     assert client.post_calls == 5
     assert [d["data"]["title"] for _, d in client.posts] == [
@@ -379,7 +385,7 @@ def test_dispatch_batches_across_buckets():
             _heartbeat(i * 10, data={"title": "same"}),
         )
 
-    rq._dispatch_request()
+    _drain(rq)
 
     assert client.post_calls == 2
     assert sorted(endpoint for endpoint, _ in client.posts) == [
@@ -432,7 +438,7 @@ def _post(self, endpoint, data, **kwargs):
     rq.add_request("buckets/bad/heartbeat?pulsetime=10", _heartbeat(0))
     rq.add_request("buckets/good/heartbeat?pulsetime=10", _heartbeat(0, data={"k": 1}))
 
-    rq._dispatch_request()
+    _drain(rq)
 
     assert client.post_calls == 2
     assert [endpoint for endpoint, _ in client.posts] == [
@@ -465,3 +471,83 @@ def test_dispatch_drains_10k_heartbeats_in_batches():
     # The first request already spans the whole backlog (no data loss).
     assert client.posts[0][1]["duration"] > 0
     assert elapsed < 10, f"draining 10k heartbeats took {elapsed:.1f}s"
+
+
+def test_dispatch_preserves_order_when_pulsetimes_differ():
+    """
+    Heartbeats for one bucket with different pulsetimes must not be reordered
+    (grouping by endpoint alone would merge/reorder across the intervening
+    one and change the server-side timeline).
+    """
+    client = RecordingClient()
+    rq = _fresh_queue(client)
+    rq.connected = True
+
+    rq.add_request("buckets/test/heartbeat?pulsetime=10", _heartbeat(0))
+    rq.add_request("buckets/test/heartbeat?pulsetime=20", _heartbeat(10))
+    rq.add_request("buckets/test/heartbeat?pulsetime=10", _heartbeat(20))
+
+    _drain(rq)
+
+    assert [endpoint for endpoint, _ in client.posts] == [
+        "buckets/test/heartbeat?pulsetime=10",
+        "buckets/test/heartbeat?pulsetime=20",
+        "buckets/test/heartbeat?pulsetime=10",
+    ]
+    assert rq._get_next() is None
+
+
+def test_dispatch_handles_malformed_pulsetime_without_crashing():
+    """A malformed pulsetime must not raise inside coalescing: the request is
+    sent verbatim and the rest of the queue still drains."""
+    client = RecordingClient()
+    rq = _fresh_queue(client)
+    rq.connected = True
+
+    rq.add_request("buckets/test/heartbeat?pulsetime=1..2", _heartbeat(0))
+    rq.add_request("buckets/test/heartbeat?pulsetime=10", _heartbeat(0, data={"k": 1}))
+
+    _drain(rq)
+
+    assert client.post_calls == 2
+    assert rq._get_next() is None
+
+
+def test_partial_retry_does_not_replay_delivered_requests():
+    """
+    If an earlier request in a batch is delivered and a later one hits a
+    transient error, the retry must resume at the failed request instead of
+    replaying the whole batch.
+    """
+
+    class FailSecondClient(RecordingClient):
+        def __init__(self):
+            super().__init__()
+            self.calls = 0
+
+        def _post(self, endpoint, data, **kwargs):
+            self.calls += 1
+            if self.calls == 2:
+                raise _http_error(503)
+            return super()._post(endpoint, data, **kwargs)
+
+    client = FailSecondClient()
+    rq = _fresh_queue(client)
+    rq.connected = True
+
+    # Different data, so these are two separate (non-mergeable) requests.
+    rq.add_request(
+        "buckets/test/heartbeat?pulsetime=10", _heartbeat(0, data={"title": "a"})
+    )
+    rq.add_request(
+        "buckets/test/heartbeat?pulsetime=10", _heartbeat(10, data={"title": "b"})
+    )
+
+    rq._dispatch_request()  # a -> delivered
+    rq._dispatch_request()  # b -> 503, stays
+    rq._dispatch_request()  # b -> delivered
+
+    assert client.calls == 3, "the failed request should be retried, not the batch"
+    assert client.post_calls == 2
+    assert [d["data"]["title"] for _, d in client.posts] == ["a", "b"]
+    assert rq._get_next() is None

From 1e7dc369a5c488f76afe7519084cc9d3be7fb7ad Mon Sep 17 00:00:00 2001
From: TimeToBuildBob <timetobuildbob@gmail.com>
Date: Fri, 2 Oct 2026 04:50:58 +0000
Subject: [PATCH 4/9] test(queue): assert the 10k drain covers the whole range;
 drop flaky timing bound

Review feedback: the 10k test only checked the first request's duration, so it
would pass even if later batches lost heartbeats; assert the summed span across
all posted events instead. Remove the wall-clock limit, which made a functional
test sensitive to runner load.

Git-Session-Id: b337
---
 tests/test_requestqueue.py | 10 +++++-----
 1 file changed, 5 insertions(+), 5 deletions(-)

diff --git a/tests/test_requestqueue.py b/tests/test_requestqueue.py
index eb525de..637e5b2 100644
--- a/tests/test_requestqueue.py
+++ b/tests/test_requestqueue.py
@@ -460,17 +460,17 @@ def test_dispatch_drains_10k_heartbeats_in_batches():
     for i in range(n):
         rq.add_request("buckets/test/heartbeat?pulsetime=10", _heartbeat(i * 10))
 
-    start = time.monotonic()
     while rq._get_next_batch():
         rq._dispatch_request()
-    elapsed = time.monotonic() - start
 
     # Every batch merges to a single heartbeat request.
     assert client.post_calls == n // rq.BATCH_SIZE
     assert rq._get_next() is None
-    # The first request already spans the whole backlog (no data loss).
-    assert client.posts[0][1]["duration"] > 0
-    assert elapsed < 10, f"draining 10k heartbeats took {elapsed:.1f}s"
+    # No data loss: the merged events across all batches still cover the whole
+    # queued range, not just the first request.
+    total = sum(data["duration"] for _, data in client.posts)
+    assert total == pytest.approx(n * 10)
+    assert all(data["data"] == {"status": "not-afk"} for _, data in client.posts)
 
 
 def test_dispatch_preserves_order_when_pulsetimes_differ():

From b22083888533c67d64947e706becba7361a662c0 Mon Sep 17 00:00:00 2001
From: Bob <bob@superuserlabs.org>
Date: Tue, 6 Oct 2026 19:23:09 +0000
Subject: [PATCH 5/9] test(queue): assert the 10k drain's merged events tile
 the queued range contiguously

Git-Session-Id: 3b002f8a-7967-5ade-8e69-6c86e374a155
---
 tests/test_requestqueue.py | 8 ++++++++
 1 file changed, 8 insertions(+)

diff --git a/tests/test_requestqueue.py b/tests/test_requestqueue.py
index 637e5b2..e8d2b91 100644
--- a/tests/test_requestqueue.py
+++ b/tests/test_requestqueue.py
@@ -471,6 +471,14 @@ def test_dispatch_drains_10k_heartbeats_in_batches():
     total = sum(data["duration"] for _, data in client.posts)
     assert total == pytest.approx(n * 10)
     assert all(data["data"] == {"status": "not-afk"} for _, data in client.posts)
+    # ...and they tile it contiguously: the first starts at the first queued
+    # heartbeat, each starts where the previous ended, the last ends at the end.
+    expected_start = _BASE
+    for _, data in client.posts:
+        start = datetime.fromisoformat(str(data["timestamp"]))
+        assert start == expected_start
+        expected_start = start + timedelta(seconds=data["duration"])
+    assert expected_start == _BASE + timedelta(seconds=n * 10)
 
 
 def test_dispatch_preserves_order_when_pulsetimes_differ():

From ca2eb486d5031196ae96499821bba5dbfa8ad3dc Mon Sep 17 00:00:00 2001
From: Bob <bob@superuserlabs.org>
Date: Tue, 6 Oct 2026 19:48:15 +0000
Subject: [PATCH 6/9] style: reformat assert to satisfy black

Git-Session-Id: 7c62adb8-40f8-58a7-82c2-170ab3f2baaa
---
 tests/test_requestqueue.py | 6 +++---
 1 file changed, 3 insertions(+), 3 deletions(-)

diff --git a/tests/test_requestqueue.py b/tests/test_requestqueue.py
index e8d2b91..cc2b68b 100644
--- a/tests/test_requestqueue.py
+++ b/tests/test_requestqueue.py
@@ -340,9 +340,9 @@ def test_dispatch_merges_consecutive_queued_heartbeats():
 
     rq._dispatch_request()
 
-    assert client.post_calls == 1, (
-        "mergeable heartbeats should collapse into one request"
-    )
+    assert (
+        client.post_calls == 1
+    ), "mergeable heartbeats should collapse into one request"
     assert rq._get_next() is None, "the queue should be fully drained"
     endpoint, data = client.posts[0]
     assert endpoint == "buckets/test/heartbeat?pulsetime=10"

From a8a5d20bb2540bf0cc2702590606116470819657 Mon Sep 17 00:00:00 2001
From: Bob <bob@superuserlabs.org>
Date: Sat, 10 Oct 2026 14:10:56 +0000
Subject: [PATCH 7/9] test(queue): adapt to master after rebase onto
 #124/#125/#130

- tests: pass tmp_path to _fresh_queue (#130 made queues isolated per test)
- RequestQueue.__init__: keep _queue_write_failing (#124) after the batch state
- wait_for_queue_empty (#125): track the in-flight batch instead of _current
- drop an import left unused by the rebase

Git-Session-Id: 68f2e145-fcff-459d-8e16-c3e43269f5fa
---
 aw_client/client.py        |  2 +-
 tests/test_requestqueue.py | 37 ++++++++++++++++++-------------------
 2 files changed, 19 insertions(+), 20 deletions(-)

diff --git a/aw_client/client.py b/aw_client/client.py
index 39f89e7..a34d1b6 100644
--- a/aw_client/client.py
+++ b/aw_client/client.py
@@ -691,13 +691,13 @@ def __init__(
         # dropping them. A single task_done() deletes every row <= the cursor,
         # i.e. the whole batch, so no per-item bookkeeping is needed.
         self._current_batch = []  # type: List[QueuedRequest]
-        self._queue_write_failing = False
         # `_current_batch` after coalescing, plus how many of those have been
         # handled. Dispatching one coalesced request per call means a transient
         # error retries only the failed request - never replays ones that
         # already reached the server - and lets the run loop observe stop().
         self._coalesced_batch = []  # type: List[QueuedRequest]
         self._coalesced_index = 0
+        self._queue_write_failing = False
 
     def _get_next(self) -> Optional[QueuedRequest]:
         # Returns the head of the in-flight batch, otherwise pops a single
diff --git a/tests/test_requestqueue.py b/tests/test_requestqueue.py
index cc2b68b..73cb8b4 100644
--- a/tests/test_requestqueue.py
+++ b/tests/test_requestqueue.py
@@ -8,7 +8,6 @@
 """
 
 import threading
-import time
 from datetime import datetime, timedelta, timezone
 from time import sleep
 from logging import basicConfig, DEBUG, WARNING
@@ -324,14 +323,14 @@ def _post(self, endpoint, data, **kwargs):
         return requests.Response()
 
 
-def test_dispatch_merges_consecutive_queued_heartbeats():
+def test_dispatch_merges_consecutive_queued_heartbeats(tmp_path):
     """
     A long offline backlog of heartbeats for one bucket must collapse into a
     handful of requests instead of one per heartbeat, without losing data
     (issue #32 / #7).
     """
     client = RecordingClient()
-    rq = _fresh_queue(client)
+    rq = _fresh_queue(client, tmp_path)
     rq.connected = True
 
     n = 200
@@ -351,10 +350,10 @@ def test_dispatch_merges_consecutive_queued_heartbeats():
     assert data["data"] == {"status": "not-afk"}
 
 
-def test_dispatch_does_not_merge_heartbeats_with_different_data():
+def test_dispatch_does_not_merge_heartbeats_with_different_data(tmp_path):
     """Heartbeats whose data differs must each be sent, in order."""
     client = RecordingClient()
-    rq = _fresh_queue(client)
+    rq = _fresh_queue(client, tmp_path)
     rq.connected = True
 
     for i in range(5):
@@ -372,10 +371,10 @@ def test_dispatch_does_not_merge_heartbeats_with_different_data():
     assert rq._get_next() is None
 
 
-def test_dispatch_batches_across_buckets():
+def test_dispatch_batches_across_buckets(tmp_path):
     """Interleaved buckets are grouped, so each bucket drains in one request."""
     client = RecordingClient()
-    rq = _fresh_queue(client)
+    rq = _fresh_queue(client, tmp_path)
     rq.connected = True
 
     for i in range(50):
@@ -395,10 +394,10 @@ def test_dispatch_batches_across_buckets():
     assert rq._get_next() is None
 
 
-def test_dispatch_retry_keeps_whole_batch_then_merges():
+def test_dispatch_retry_keeps_whole_batch_then_merges(tmp_path):
     """A transient error retains the whole batch; the retry merges it."""
     client = FlakyClient(_http_error(503))
-    rq = _fresh_queue(client)
+    rq = _fresh_queue(client, tmp_path)
     rq.connected = True
 
     n = 50
@@ -417,7 +416,7 @@ def test_dispatch_retry_keeps_whole_batch_then_merges():
     assert rq._get_next() is None
 
 
-def test_dispatch_drops_bad_request_but_keeps_the_rest():
+def test_dispatch_drops_bad_request_but_keeps_the_rest(tmp_path):
     """
     A non-retryable error (HTTP 400) drops only that request; the rest of the
     batch must still be dispatched and the queue fully drained.
@@ -432,7 +431,7 @@ def _post(self, endpoint, data, **kwargs):
             return requests.Response()
 
     client = SelectiveFailClient()
-    rq = _fresh_queue(client)
+    rq = _fresh_queue(client, tmp_path)
     rq.connected = True
 
     rq.add_request("buckets/bad/heartbeat?pulsetime=10", _heartbeat(0))
@@ -447,13 +446,13 @@ def _post(self, endpoint, data, **kwargs):
     assert rq._get_next() is None
 
 
-def test_dispatch_drains_10k_heartbeats_in_batches():
+def test_dispatch_drains_10k_heartbeats_in_batches(tmp_path):
     """
     A large offline backlog (10k mergeable heartbeats) must drain with one
     request per batch, quickly and without losing data.
     """
     client = RecordingClient()
-    rq = _fresh_queue(client)
+    rq = _fresh_queue(client, tmp_path)
     rq.connected = True
 
     n = 10_000
@@ -481,14 +480,14 @@ def test_dispatch_drains_10k_heartbeats_in_batches():
     assert expected_start == _BASE + timedelta(seconds=n * 10)
 
 
-def test_dispatch_preserves_order_when_pulsetimes_differ():
+def test_dispatch_preserves_order_when_pulsetimes_differ(tmp_path):
     """
     Heartbeats for one bucket with different pulsetimes must not be reordered
     (grouping by endpoint alone would merge/reorder across the intervening
     one and change the server-side timeline).
     """
     client = RecordingClient()
-    rq = _fresh_queue(client)
+    rq = _fresh_queue(client, tmp_path)
     rq.connected = True
 
     rq.add_request("buckets/test/heartbeat?pulsetime=10", _heartbeat(0))
@@ -505,11 +504,11 @@ def test_dispatch_preserves_order_when_pulsetimes_differ():
     assert rq._get_next() is None
 
 
-def test_dispatch_handles_malformed_pulsetime_without_crashing():
+def test_dispatch_handles_malformed_pulsetime_without_crashing(tmp_path):
     """A malformed pulsetime must not raise inside coalescing: the request is
     sent verbatim and the rest of the queue still drains."""
     client = RecordingClient()
-    rq = _fresh_queue(client)
+    rq = _fresh_queue(client, tmp_path)
     rq.connected = True
 
     rq.add_request("buckets/test/heartbeat?pulsetime=1..2", _heartbeat(0))
@@ -521,7 +520,7 @@ def test_dispatch_handles_malformed_pulsetime_without_crashing():
     assert rq._get_next() is None
 
 
-def test_partial_retry_does_not_replay_delivered_requests():
+def test_partial_retry_does_not_replay_delivered_requests(tmp_path):
     """
     If an earlier request in a batch is delivered and a later one hits a
     transient error, the retry must resume at the failed request instead of
@@ -540,7 +539,7 @@ def _post(self, endpoint, data, **kwargs):
             return super()._post(endpoint, data, **kwargs)
 
     client = FailSecondClient()
-    rq = _fresh_queue(client)
+    rq = _fresh_queue(client, tmp_path)
     rq.connected = True
 
     # Different data, so these are two separate (non-mergeable) requests.

From bfa8663ddedc968f8b428c975d5bff552090b880 Mon Sep 17 00:00:00 2001
From: Bob <bob@superuserlabs.org>
Date: Wed, 14 Oct 2026 11:39:31 +0000
Subject: [PATCH 8/9] fix(queue): preserve ordering and survive malformed event
 values

Git-Session-Id: dac9f7d8-b49f-543c-8deb-b406c61fe5e2
---
 aw_client/client.py        | 13 +++++---
 tests/test_requestqueue.py | 65 ++++++++++++++++++++++++++++++++++++--
 2 files changed, 71 insertions(+), 7 deletions(-)

diff --git a/aw_client/client.py b/aw_client/client.py
index a34d1b6..3b664a4 100644
--- a/aw_client/client.py
+++ b/aw_client/client.py
@@ -3,6 +3,7 @@
 import ipaddress
 import json
 import logging
+import math
 import os
 import re
 import socket
@@ -597,7 +598,7 @@ def _retry_delay(response: req.Response) -> float:
 # Matches the endpoint the queue stores heartbeats under, capturing the bucket
 # id and the pulsetime needed to merge consecutive queued heartbeats.
 _HEARTBEAT_ENDPOINT_RE = re.compile(
-    r"^buckets/(?P<bucket_id>.+)/heartbeat\?pulsetime=(?P<pulsetime>[0-9]+(?:\.[0-9]+)?)$"
+    r"^buckets/(?P<bucket_id>.+)/heartbeat\?pulsetime=(?P<pulsetime>[^&]*)$"
 )
 
 
@@ -614,6 +615,8 @@ def _parse_heartbeat_endpoint(endpoint: str) -> Optional[Tuple[str, float]]:
         pulsetime = float(match.group("pulsetime"))
     except ValueError:
         return None
+    if not math.isfinite(pulsetime) or pulsetime < 0:
+        return None
     return match.group("bucket_id"), pulsetime
 
 
@@ -623,7 +626,7 @@ def _try_event(data: Any) -> Optional[Event]:
         return None
     try:
         return Event(**data)
-    except TypeError:
+    except (TypeError, ValueError, OverflowError, AttributeError):
         return None
 
 
@@ -792,8 +795,10 @@ def _coalesce(self, batch: List[QueuedRequest]) -> List[QueuedRequest]:
         """
         groups = {}  # type: Dict[str, List[QueuedRequest]]
         for request in batch:
-            parsed = _parse_heartbeat_endpoint(request.endpoint)
-            key = parsed[0] if parsed is not None else request.endpoint
+            # Bucket identity does not depend on whether pulsetime is valid:
+            # an unmergeable request must still interrupt that bucket's run.
+            match = _HEARTBEAT_ENDPOINT_RE.match(request.endpoint)
+            key = match.group("bucket_id") if match is not None else request.endpoint
             groups.setdefault(key, []).append(request)
 
         coalesced = []  # type: List[QueuedRequest]
diff --git a/tests/test_requestqueue.py b/tests/test_requestqueue.py
index 73cb8b4..0b37380 100644
--- a/tests/test_requestqueue.py
+++ b/tests/test_requestqueue.py
@@ -7,6 +7,7 @@
 with confidence, and I need some of that right now.
 """
 
+import json
 import threading
 from datetime import datetime, timedelta, timezone
 from time import sleep
@@ -339,9 +340,9 @@ def test_dispatch_merges_consecutive_queued_heartbeats(tmp_path):
 
     rq._dispatch_request()
 
-    assert (
-        client.post_calls == 1
-    ), "mergeable heartbeats should collapse into one request"
+    assert client.post_calls == 1, (
+        "mergeable heartbeats should collapse into one request"
+    )
     assert rq._get_next() is None, "the queue should be fully drained"
     endpoint, data = client.posts[0]
     assert endpoint == "buckets/test/heartbeat?pulsetime=10"
@@ -558,3 +559,61 @@ def _post(self, endpoint, data, **kwargs):
     assert client.post_calls == 2
     assert [d["data"]["title"] for _, d in client.posts] == ["a", "b"]
     assert rq._get_next() is None
+
+
+@pytest.mark.parametrize("pulsetime", ["1e-05", "1E+1", "1..2"])
+def test_dispatch_preserves_order_around_unusual_pulsetimes(tmp_path, pulsetime):
+    """Parsing failure must not split one bucket into reordered groups."""
+    client = RecordingClient()
+    rq = _fresh_queue(client, tmp_path)
+    endpoints = [
+        "buckets/test/heartbeat?pulsetime=10",
+        f"buckets/test/heartbeat?pulsetime={pulsetime}",
+        "buckets/test/heartbeat?pulsetime=10",
+    ]
+    for i, endpoint in enumerate(endpoints):
+        rq.add_request(endpoint, _heartbeat(i * 10))
+
+    _drain(rq)
+
+    assert [endpoint for endpoint, _ in client.posts] == endpoints
+    assert [data["timestamp"] for _, data in client.posts] == [
+        _heartbeat(i * 10)["timestamp"] for i in range(3)
+    ]
+    assert rq.wait_for_queue_empty(timeout=0)
+
+
+@pytest.mark.parametrize(
+    "invalid_fields",
+    [
+        {"timestamp": "not-a-timestamp"},
+        {"timestamp": 123},
+        {"duration": float("nan")},
+        {"duration": float("inf")},
+    ],
+)
+def test_dispatch_invalid_event_does_not_block_later_heartbeats(
+    tmp_path, invalid_fields
+):
+    """Bad event values go to dispatch unchanged instead of killing coalescing."""
+    client = RecordingClient()
+    rq = _fresh_queue(client, tmp_path)
+    endpoint = "buckets/test/heartbeat?pulsetime=10"
+    invalid = dict(_heartbeat(10), **invalid_fields)
+    payloads = [_heartbeat(0), invalid, _heartbeat(20)]
+    for payload in payloads:
+        rq.add_request(endpoint, payload)
+
+    _drain(rq)
+
+    assert len(client.posts) == 3
+    assert client.posts[1][0] == endpoint
+    assert json.dumps(client.posts[1][1], sort_keys=True) == json.dumps(
+        invalid, sort_keys=True
+    )
+    for index in (0, 2):
+        assert all(
+            client.posts[index][1][key] == value
+            for key, value in payloads[index].items()
+        )
+    assert rq.wait_for_queue_empty(timeout=0)

From b539af579737ddf172ae0b38c2bccc146f6cce2e Mon Sep 17 00:00:00 2001
From: Bob <bob@superuserlabs.org>
Date: Wed, 14 Oct 2026 11:41:38 +0000
Subject: [PATCH 9/9] style(queue): match CI black formatting

Git-Session-Id: dac9f7d8-b49f-543c-8deb-b406c61fe5e2
---
 tests/test_requestqueue.py | 6 +++---
 1 file changed, 3 insertions(+), 3 deletions(-)

diff --git a/tests/test_requestqueue.py b/tests/test_requestqueue.py
index 0b37380..0ddc42c 100644
--- a/tests/test_requestqueue.py
+++ b/tests/test_requestqueue.py
@@ -340,9 +340,9 @@ def test_dispatch_merges_consecutive_queued_heartbeats(tmp_path):
 
     rq._dispatch_request()
 
-    assert client.post_calls == 1, (
-        "mergeable heartbeats should collapse into one request"
-    )
+    assert (
+        client.post_calls == 1
+    ), "mergeable heartbeats should collapse into one request"
     assert rq._get_next() is None, "the queue should be fully drained"
     endpoint, data = client.posts[0]
     assert endpoint == "buckets/test/heartbeat?pulsetime=10"