diff --git a/src/agentcompile/_capture.py b/src/agentcompile/_capture.py index ae03b79..ddc6c6c 100644 --- a/src/agentcompile/_capture.py +++ b/src/agentcompile/_capture.py @@ -179,13 +179,22 @@ def _take(self) -> list[str]: self.dropped += 1 continue if batch and size + len(line) > MAX_BATCH_BYTES: - with self._lock: # the rest waits for the next batch, in order - self._queue.extendleft(reversed(records[i:])) + self._put_back(records[i:]) break batch.append(line) size += len(line) return batch + def _put_back(self, rest: list[dict[str, Any]]) -> None: + """Calls taken for a batch that didn't fit, back at the front, in order. They are the + oldest: when newer calls filled the queue meanwhile, they are the ones dropped, and + counted (extendleft would evict the newest instead, silently).""" + with self._lock: + room = (self._queue.maxlen or len(rest)) - len(self._queue) + keep = rest[len(rest) - max(room, 0) :] + self.dropped += len(rest) - len(keep) + self._queue.extendleft(reversed(keep)) + def _send_batch(self) -> None: batch = self._take() if not batch: diff --git a/tests/test_hardening.py b/tests/test_hardening.py index 7b38712..ab3fe96 100644 --- a/tests/test_hardening.py +++ b/tests/test_hardening.py @@ -312,3 +312,14 @@ def test_wrapping_again_reuses_the_sender_and_keeps_one_destination(tmp_path: Pa ) assert len(_capture._ALL) == 1 assert len(_outcome._settings) == 1 + + +def test_putting_a_batch_back_on_a_full_queue_drops_the_oldest_counted() -> None: + import collections + + capturer = _capture.Capturer(Settings("http://ac.test", "ack_acme.k", None, 1.0)) + capturer._queue = collections.deque([{"id": n} for n in (4, 5, 6)], maxlen=4) + capturer._put_back([{"id": 2}, {"id": 3}]) # older calls, one free place + assert [r["id"] for r in capturer._queue] == [3, 4, 5, 6] and capturer.dropped == 1 + capturer._put_back([{"id": 1}]) # no room: the oldest is the one dropped + assert [r["id"] for r in capturer._queue] == [3, 4, 5, 6] and capturer.dropped == 2