diff --git a/.gitignore b/.gitignore
index 424ea99..210e8ae 100644
--- a/.gitignore
+++ b/.gitignore
@@ -24,3 +24,6 @@ docs/
 .claude/
 .agents/
 skills-lock.json
+
+# logs
+*.log
diff --git a/CHANGELOG.md b/CHANGELOG.md
index fbe523d..404b019 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -5,6 +5,68 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [0.1.4] - 2026-08-17
+
+### Added
+
+- New `rate_limited` counter in `HandlerStats`: 429 `Retry-After` waits are
+  now counted even though they never consume the retry budget, so a
+  rate-limit storm shows up in the stats instead of reporting zero retries.
+
+### Fixed
+
+- Failed handler construction no longer leaves a half-built handler registered
+  with the logging module. Previously, exit-time `logging.shutdown()` called
+  `close()` on the half-built instance and printed a stray `AttributeError`
+  after the real construction error; the registration is now dropped on
+  failure. The cleanup itself is failure-proof (`contextlib.suppress`), so a
+  teardown error (e.g. logging internals already torn down at interpreter
+  shutdown) can never mask the original construction error.
+- Live smoke tests no longer race the worker: `close()` returns once the
+  `shutdown_timeout` elapses even while the daemon worker is still sending its
+  last batches, so the tests now poll the stats until the counters converge
+  before asserting. A new live test proves 10 records with `batch_size=5` go
+  out as exactly two 5-record messages (`batches_sent == 2`).
+- Worker shutdown is event-backed: a saturated queue or a `drop_oldest`
+  eviction can no longer lose the shutdown sentinel and leak the worker
+  thread (and its HTTP client) after `close()`.
+- **Shutdown no longer waits out a long `flush_interval`**: the accumulator
+  polls the shutdown event between capped waits, so `close()` returns promptly
+  (≤ ~1s) even with a partial batch pending and the sentinel lost; the pending
+  batch is still flushed first.
+- A close() that interleaves with an in-flight `emit()` no longer orphans the
+  record: `emit` and `close` are mutually exclusive, so nothing accepted
+  before `close()` is lost (and `queue_full_policy="block"` can no longer hang
+  an emitter past shutdown).
+- Post-close `emit()` drops and counts the record instead of potentially
+  hanging the caller under `queue_full_policy="block"`.
+- `truncate_text` clamps the marker when the limit is smaller than the marker
+  itself, so a truncated message can never exceed the cap (Telegram 400s
+  oversized payloads).
+- The one-time 429 heads-up notification now swallows any failure (not just
+  transport errors), so it can never surface through `send_with_retry`.
+- **Batches are no longer split by a pending shutdown.** The shutdown poll now
+  runs only after a capped wait elapsed with nothing arriving, so records that
+  keep flowing still group into one batch even when `close()` was called during
+  the drain (previously every remaining record flushed as its own message).
+- A non-200 `sendMessage` response (a 3xx — redirects are never followed — or
+  a non-200 2xx) is no longer reported as delivered; getMe validation
+  maps non-JSON 200 bodies to `TelegramConfigError` instead of a bare
+  `ValueError`; env `TG_CHAT_ID` trailing whitespace is stripped like the
+  token's; queue `task_done` accounting is now exact, so `queue.join()` can
+  no longer hang.
+- `close()` now enqueues the shutdown sentinel **before** signaling the
+  worker, so the worker can never exit while the sentinel is still about to be
+  queued and strand it in the queue (a user's `queue.join()` would hang).
+- getMe validation requires an object body with `ok == True` exactly: a
+  scalar/list JSON body or a merely truthy `ok` (e.g. `1`) maps to
+  `TelegramConfigError` instead of a raw `AttributeError` or a false accept.
+- The clamped truncation marker is nudged off a trailing backslash, so a tiny
+  `max_length` with MarkdownV2 can never leave a dangling escape pair.
+- HTTP redirects are disabled on the send client: a 302/308 with a `Location`
+  is never followed, so a redirected `sendMessage` request can never be
+  mistaken for a delivery.
+
 ## [0.1.3] - 2026-08-16
 
 ### Added
diff --git a/localtest.bat b/localtest.bat
new file mode 100644
index 0000000..decb3f7
--- /dev/null
+++ b/localtest.bat
@@ -0,0 +1,9 @@
+@echo off
+(
+  uv run mypy || exit /b 1
+  uv run ruff format --check . || exit /b 1
+  uv run ruff check . || exit /b 1
+  uv run pylint tests tg_logging_handler || exit /b 1
+  uv run flake8 tests tg_logging_handler || exit /b 1
+  uv run pytest --cov=tg_logging_handler || exit /b 1
+) > local.log 2>&1
diff --git a/pyproject.toml b/pyproject.toml
index 08e143c..3b0492c 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "tg-logging-handler"
-version = "0.1.3"
+version = "0.1.4"
 description = "Production-grade Telegram destination for Python's standard logging module"
 readme = "README.md"
 requires-python = ">=3.10"
diff --git a/tests/test_batching.py b/tests/test_batching.py
index 27fc0d7..bb7bdfc 100644
--- a/tests/test_batching.py
+++ b/tests/test_batching.py
@@ -109,3 +109,68 @@ def test_constructor_validates_args() -> None:
         BatchAccumulator(0, 1.0)
     with pytest.raises(ValueError):
         BatchAccumulator(1, -1.0)
+
+
+class _TimeoutQueue:
+    """Scripted queue whose get() always times out (raises Empty)."""
+
+    def __init__(self) -> None:
+        self._items: list[object] = []
+
+    def put_nowait(self, item: object) -> None:
+        self._items.append(item)
+
+    def get(self, timeout: float | None = None) -> object:
+        if not self._items:
+            raise queue.Empty
+        return self._items.pop(0)
+
+    def task_done(self) -> None:
+        pass
+
+
+def test_second_get_timeout_returns_partial_batch() -> None:
+    # The mid-batch get() can time out with nothing new arriving; the partial
+    # batch must be returned, not lost. Constant clock keeps remaining > 0 so
+    # the get() is actually attempted (and times out immediately, scripted).
+    acc, _ = _mk(batch_size=5, flush_interval=1.0, clock=ScriptedClock(0.0, 0.0))
+    q = _TimeoutQueue()
+    q.put_nowait(_record("only"))
+    batch, shutdown = acc.collect(q, Shutdown, timeout=0)  # type: ignore[arg-type]
+    assert [r.getMessage() for r in batch] == ["only"]
+    assert not shutdown
+
+
+def test_stop_flushes_partial_batch_promptly() -> None:
+    # A shutdown requested while a partial batch is mid-collection must flush
+    # it immediately, not wait out the flush interval (potentially minutes);
+    # collect polls the stop predicate after a capped wait with nothing
+    # arriving.
+    acc, q = _mk(batch_size=10, flush_interval=30.0)
+    _put(q, "only")
+    batch, shutdown = acc.collect(q, Shutdown, timeout=0, stop=lambda: True)
+    assert [r.getMessage() for r in batch] == ["only"]
+    assert shutdown is False
+
+
+def test_stop_does_not_split_a_flowing_batch() -> None:
+    # Live-smoke regression: close() sets the shutdown event while the worker
+    # is still draining a queue full of records. stop must only be consulted
+    # once a wait actually elapsed with nothing arriving, or every flowing
+    # record flushes as its own batch. stop=True here must not split the 5
+    # queued records (batch_size is exactly 5, so they must come back whole).
+    acc, q = _mk(batch_size=5, flush_interval=10.0)
+    _put(q, "a", "b", "c", "d", "e")
+    batch, shutdown = acc.collect(q, Shutdown, timeout=0, stop=lambda: True)
+    assert [r.getMessage() for r in batch] == ["a", "b", "c", "d", "e"]
+    assert not shutdown
+
+
+def test_stop_false_after_capped_wait_keeps_collecting() -> None:
+    # Stop present but still False when a capped wait elapses: the loop must
+    # keep going (re-check the interval) instead of aborting an active batch.
+    acc, q = _mk(batch_size=10, flush_interval=5.0, clock=ScriptedClock(0.0, 0.0, 0.0, 5.0))
+    _put(q, "a", "b")
+    batch, shutdown = acc.collect(q, Shutdown, timeout=0, stop=lambda: False)
+    assert [r.getMessage() for r in batch] == ["a", "b"]
+    assert not shutdown
diff --git a/tests/test_config_stats.py b/tests/test_config_stats.py
index 2ea16ff..aabd59c 100644
--- a/tests/test_config_stats.py
+++ b/tests/test_config_stats.py
@@ -52,6 +52,14 @@ def test_trailing_newline_token_is_stripped(self) -> None:
         # must not fail validation or leak into the request URL.
         assert resolve_credentials("111:abc\n", 1) == ("111:abc", "1")
 
+    def test_env_chat_id_trailing_newline_is_stripped(
+        self, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        # Same story for chats: .env files routinely end values with a
+        # newline, and an unstripped one would 400 at send time.
+        monkeypatch.setenv("TG_CHAT_ID", "-42\n")
+        assert resolve_credentials("111:abc", None)[1] == "-42"
+
     def test_invalid_short_token_error_is_redacted(self) -> None:
         # A token too short for a safe partial reveal must not appear in the
         # error message at all.
@@ -163,6 +171,73 @@ def get_nowait(self) -> object:
             raise queue.Empty
         return "evicted"
 
+    def task_done(self) -> None:
+        pass
+
+
+class TestValidateToken:
+    def test_non_json_200_body_is_config_error(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        # A proxy/gateway can answer 200 with an HTML error page; the docs
+        # promise TelegramConfigError, never a bare ValueError.
+        import httpx
+
+        from tg_logging_handler.config import validate_token
+
+        monkeypatch.setattr(
+            httpx,
+            "get",
+            lambda *args, **kwargs: httpx.Response(200, text="<html>gateway error</html>"),
+        )
+        with pytest.raises(TelegramConfigError, match="non-JSON"):
+            validate_token("111:abc", "https://api.telegram.org")
+
+    def test_network_error_is_config_error(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        # An unreachable API (offline host, DNS failure) must surface as the
+        # documented TelegramConfigError, with the offline-escape hatch hint.
+        import httpx
+
+        from tg_logging_handler.config import validate_token
+
+        def offline(*args: object, **kwargs: object) -> httpx.Response:
+            raise httpx.ConnectError("offline")
+
+        monkeypatch.setattr(httpx, "get", offline)
+        with pytest.raises(TelegramConfigError, match="Could not reach Telegram"):
+            validate_token("111:abc", "https://api.telegram.org")
+
+    @pytest.mark.parametrize("payload", [[], "hello", 42])
+    def test_non_object_json_200_body_is_config_error(
+        self, monkeypatch: pytest.MonkeyPatch, payload: object
+    ) -> None:
+        # A 200 that decodes as a scalar/list (never an object) cannot carry
+        # ok; it must hit the TelegramConfigError path, not an AttributeError.
+        import httpx
+
+        from tg_logging_handler.config import validate_token
+
+        monkeypatch.setattr(
+            httpx,
+            "get",
+            lambda *args, **kwargs: httpx.Response(200, json=payload),
+        )
+        with pytest.raises(TelegramConfigError, match="rejected"):
+            validate_token("111:abc", "https://api.telegram.org")
+
+    def test_truthy_but_not_true_ok_is_config_error(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        # ok == 1 is truthy but not exactly True; only a boolean True means
+        # the token was accepted.
+        import httpx
+
+        from tg_logging_handler.config import validate_token
+
+        monkeypatch.setattr(
+            httpx,
+            "get",
+            lambda *args, **kwargs: httpx.Response(200, json={"ok": 1}),
+        )
+        with pytest.raises(TelegramConfigError, match="rejected"):
+            validate_token("111:abc", "https://api.telegram.org")
+
 
 class TestQueuePolicy:
     def test_put_drop_newest_succeeds_when_space(self) -> None:
@@ -190,6 +265,16 @@ def test_put_drop_oldest_evicts_to_make_room(self) -> None:
         assert result.dropped == 1  # evicted the oldest (1)
         assert list(q.queue) == [2, 3]
 
+    def test_put_drop_oldest_task_dones_evicted_items(self) -> None:
+        # Every eviction is a get(); without a matching task_done(),
+        # queue.join() would hang forever on a full-and-drained queue.
+        q: queue.Queue[object] = queue.Queue(maxsize=2)
+        q.put_nowait(1)
+        q.put_nowait(2)
+        queue_policy.put_drop_oldest(q, 3)
+        # Only items still in the queue may lack a task_done().
+        assert q.unfinished_tasks == q.qsize() == 2
+
     def test_put_drop_oldest_no_eviction_when_space(self) -> None:
         q: queue.Queue[object] = queue.Queue(maxsize=2)
         q.put_nowait(1)
diff --git a/tests/test_handler.py b/tests/test_handler.py
index bf5f9f5..679d8f4 100644
--- a/tests/test_handler.py
+++ b/tests/test_handler.py
@@ -3,9 +3,11 @@
 from __future__ import annotations
 
 import logging
+import queue
 import threading
 import time
 from collections.abc import Callable
+from typing import Any, cast
 
 import httpx
 import pytest
@@ -19,6 +21,7 @@
     TGLoggingHandler,
     __all__,
 )
+from tg_logging_handler.queue_policy import PutResult
 
 HandlerFactory = Callable[..., TelegramLoggingHandler]
 WaitFor = Callable[..., bool]
@@ -54,6 +57,23 @@ def test_invalid_token_format_raises() -> None:
         TelegramLoggingHandler(token="garbage", chat_id=1, validate=False)
 
 
+def test_failed_construction_unregisters_from_handler_list(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """A half-built handler must not stay registered for exit-time shutdown()."""
+    monkeypatch.delenv("TG_TOKEN", raising=False)
+    monkeypatch.delenv("TG_CHAT_ID", raising=False)
+
+    def alive() -> int:
+        handler_list = cast(Any, logging)._handlerList
+        return sum(1 for ref in handler_list if isinstance(ref(), TelegramLoggingHandler))
+
+    before = alive()
+    with pytest.raises(TelegramConfigError):
+        TelegramLoggingHandler(validate=False)
+    assert alive() == before
+
+
 def test_env_fallback_constructs(monkeypatch: pytest.MonkeyPatch) -> None:
     monkeypatch.setenv("TG_TOKEN", TEST_TOKEN)
     monkeypatch.setenv("TG_CHAT_ID", TEST_CHAT_ID)
@@ -248,3 +268,189 @@ def emit_batch(worker_id: int) -> None:
     assert wait_for(lambda: handler.stats.sent == 2000)
     stats = handler.stats
     assert stats.sent + stats.dropped + stats.failed == 2000
+
+
+def test_close_stops_worker_even_when_queue_was_saturated(
+    fast_handler_factory: HandlerFactory, mock_api: respx.MockRouter, wait_for: WaitFor
+) -> None:
+    # Regression: with a saturated queue the SHUTDOWN put is suppressed and
+    # the worker must still terminate (event-backed shutdown); previously it
+    # leaked the thread and the httpx client forever.
+    handler = fast_handler_factory(queue_maxsize=1, flush_interval=5.0, shutdown_timeout=1.0)
+    handler.emit(_record("first"))
+    handler.emit(_record("second"))  # fills the queue: sentinel put is suppressed
+    handler.close()
+    assert wait_for(lambda: not handler._worker.is_alive())
+
+
+def test_emit_after_close_drops_instead_of_blocking(
+    mock_api: respx.MockRouter, wait_for: WaitFor
+) -> None:
+    # Regression: with queue_full_policy="block" a post-close emit on a full
+    # queue hung the calling thread forever (no worker drains it). Now it is
+    # dropped and counted.
+    handler = TelegramLoggingHandler(
+        token=TEST_TOKEN,
+        chat_id=TEST_CHAT_ID,
+        validate=False,
+        queue_maxsize=1,
+        queue_full_policy="block",
+        flush_interval=30.0,
+        shutdown_timeout=1.0,
+        api_base_url=API_BASE,
+    )
+    handler.close()
+    assert wait_for(lambda: not handler._worker.is_alive())
+    before = handler.stats.dropped
+    handler.emit(_record("post-close"))
+    assert handler.stats.dropped == before + 1
+
+
+def test_close_waits_for_in_flight_emit(mock_api: respx.MockRouter) -> None:
+    # Regression: close() once interleaved its _closed write between an emit's
+    # guard check and its enqueue, orphaning the record (counted queued but
+    # never sent; block policy: hung forever). The close lock serializes emit
+    # vs close: close() blocks until the in-flight emit has fully enqueued,
+    # so nothing accepted before close() is lost.
+    handler = TelegramLoggingHandler(
+        token=TEST_TOKEN,
+        chat_id=TEST_CHAT_ID,
+        validate=False,
+        queue_maxsize=2,
+        queue_full_policy="block",
+        flush_interval=5.0,
+        shutdown_timeout=2.0,
+        api_base_url=API_BASE,
+    )
+    started = threading.Event()
+    release = threading.Event()
+    original = handler._put_policy
+
+    def slow_policy(q: queue.Queue[object], item: object) -> PutResult:
+        result = original(q, item)
+        started.set()  # emit is now inside the put (holding the close lock)
+        release.wait(timeout=5.0)
+        return result
+
+    handler._put_policy = slow_policy
+
+    def do_emit() -> None:
+        handler.emit(_record("z"))
+
+    emitter = threading.Thread(target=do_emit)
+    emitter.start()
+    assert started.wait(timeout=2.0)
+
+    outcome: dict[str, float] = {}
+
+    def do_close() -> None:
+        t0 = time.monotonic()
+        handler.close()
+        outcome["elapsed"] = time.monotonic() - t0
+
+    closer = threading.Thread(target=do_close)
+    closer.start()
+    time.sleep(0.2)
+    assert "elapsed" not in outcome  # close() must wait for the in-flight emit
+    release.set()
+    emitter.join(timeout=2.0)
+    closer.join(timeout=2.0)
+    assert outcome["elapsed"] < 1.0  # close finished promptly once the emit did
+    assert not handler._worker.is_alive()
+    stats = handler.stats
+    assert stats.queued == 1
+    assert stats.sent == 1  # the in-flight record was drained, not orphaned
+
+
+def test_emit_swallows_put_policy_exceptions(
+    fast_handler_factory: HandlerFactory,
+) -> None:
+    # A policy that raises must never propagate out of emit (stdlib contract).
+    handler = fast_handler_factory()
+
+    def broken_policy(q: queue.Queue[object], item: object) -> PutResult:
+        raise RuntimeError("policy exploded")
+
+    handler._put_policy = broken_policy
+    handler.emit(_record("x"))  # must not raise
+    handler.close()
+
+
+def test_emit_racing_close_drops_instead_of_orphaning(
+    fast_handler_factory: HandlerFactory,
+) -> None:
+    # An emit whose snapshot was already taken when close() completed must hit
+    # the authoritative closed check under the lock and drop+count instead of
+    # enqueueing into a drained pipeline (counted queued, never sent).
+    handler = fast_handler_factory()
+    release = threading.Event()
+
+    class SlowStr:
+        def __str__(self) -> str:
+            release.wait(timeout=2.0)
+            return "boom"
+
+    slow = SlowStr()
+    # getMessage() formats msg % args, invoking SlowStr.__str__ -> the emit
+    # pauses between the fast-path guard and the close-lock check.
+    slow_record = logging.LogRecord("t", logging.ERROR, __file__, 1, "%s", (slow,), None)
+    emitter = threading.Thread(target=handler.emit, args=(slow_record,))
+    emitter.start()
+    time.sleep(0.1)  # emit is now inside getMessage, before the lock
+    handler.close()  # real close(): the emit is mid-flight past its snapshot
+    release.set()
+    emitter.join(timeout=2.0)
+    assert handler.stats.dropped == 1  # counted, not orphaned
+    handler.close()  # idempotent no-op
+
+
+def test_close_enqueues_sentinel_before_signaling_worker_shutdown(
+    fast_handler_factory: HandlerFactory,
+) -> None:
+    # close() must put the SHUTDOWN sentinel before setting the shutdown event:
+    # the worker's exit check is "event set AND queue empty", so signaling
+    # first would let it return while the sentinel is still about to be
+    # enqueued, stranding it in a queue nobody drains (queue.join() hangs).
+    handler = fast_handler_factory()
+    order: list[str] = []
+
+    worker = handler._worker
+    orig_shutdown = worker.shutdown
+
+    def spy_shutdown() -> None:
+        order.append("event")
+        orig_shutdown()
+
+    worker.shutdown = spy_shutdown  # type: ignore[method-assign]
+
+    q = handler._queue
+    orig_put = q.put_nowait
+
+    def spy_put(item: object) -> None:
+        order.append("sentinel")
+        orig_put(item)
+
+    q.put_nowait = spy_put  # type: ignore[method-assign]
+
+    handler.close()
+    assert order == ["sentinel", "event"]  # sentinel queued before the wakeup
+
+
+def test_close_with_empty_queue_strands_nothing(
+    fast_handler_factory: HandlerFactory,
+) -> None:
+    # With nothing queued, close() must still leave the queue joinable: the
+    # sentinel is either consumed by the worker or never enqueued after an
+    # exit, so queue.join() always completes.
+    handler = fast_handler_factory()
+    handler.close()
+    done = threading.Event()
+
+    def join() -> None:
+        handler._queue.join()
+        done.set()
+
+    joiner = threading.Thread(target=join)
+    joiner.start()
+    joiner.join(timeout=2.0)
+    assert done.is_set()  # join() completed; no stranded sentinel
diff --git a/tests/test_integration.py b/tests/test_integration.py
index 6485920..36a0323 100644
--- a/tests/test_integration.py
+++ b/tests/test_integration.py
@@ -97,10 +97,12 @@ def test_burst_of_100_is_batched(
         logger.error("msg-%d", i)
     route = next(r for r in mock_api.routes if "sendMessage" in str(r.pattern))
     assert wait_for(lambda: route.call_count == 10)
+    # Counters update right after the wire call, so wait for the final count
+    # instead of reading stats in the same instant as the 10th call.
+    assert wait_for(lambda: handler.stats.sent == 100)
     logger.handlers.clear()
 
     # Every record delivered, none lost, and each wire message carries 10 lines.
-    assert handler.stats.sent == 100
     assert handler.stats.batches_sent == 10
     assert all(text.count("\n") == 9 for text in _batch_texts(route))
 
diff --git a/tests/test_overflow.py b/tests/test_overflow.py
index bd96fc3..3cd3551 100644
--- a/tests/test_overflow.py
+++ b/tests/test_overflow.py
@@ -10,7 +10,7 @@
 import pytest
 
 from tg_logging_handler._constants import TELEGRAM_MAX_MESSAGE_LENGTH
-from tg_logging_handler.overflow import prepare_messages, split_text, truncate_text
+from tg_logging_handler.overflow import _safe_cut, prepare_messages, split_text, truncate_text
 
 
 def _strip_header(part: str) -> str:
@@ -132,3 +132,51 @@ def test_truncate_escapes_its_marker_for_markdownv2() -> None:
     out = truncate_text("w" * 500, 100, parse_mode="MarkdownV2")
     assert out.endswith(escape("\n… [truncated]", "MarkdownV2"))
     assert len(out) <= 100
+
+
+def test_safe_cut_odd_backslash_run_backs_up_one() -> None:
+    # A cut right after a lone backslash would split the MarkdownV2 \X pair;
+    # the cut must move left by one so the pair moves whole to the next part.
+    assert _safe_cut("\\a", 1) == 0
+
+
+def test_safe_cut_even_backslash_run_keeps_cut() -> None:
+    # Two backslashes are a literal backslash (balanced); the cut is fine.
+    assert _safe_cut("\\a", 2) == 2
+
+
+def test_safe_cut_odd_run_respects_floor() -> None:
+    # Backing up would cross the floor (no forward progress): keep the cut.
+    # r"\a" is a backslash followed by "a" — a plain "\a" would be the bell
+    # character and never exercise the backslash run.
+    assert _safe_cut(r"\a", 1, floor=1) == 1
+
+
+def test_split_keeps_markdown_v2_escape_pairs_whole() -> None:
+    # A leading character shifts every escape pair to an odd offset, so hard
+    # cuts land inside a \_ pair; every pair must stay intact, parts
+    # reconstruct byte-for-byte, and no body may end mid-escape.
+    text = "x" + r"\_" * 200
+    parts = split_text(text, 100, parse_mode="MarkdownV2")
+    bodies = [_strip_header(p) for p in parts]
+    assert "".join(bodies) == text
+    assert all(_trailing_backslashes(b) % 2 == 0 for b in bodies)
+
+
+def test_truncate_with_tiny_limit_stays_within_limit() -> None:
+    # A limit smaller than the marker itself must still clamp the result to
+    # the cap; Telegram rejects oversized payloads with a 400.
+    assert truncate_text("x" * 100, 5) == "\n… [truncated]"[:5]
+    for limit in (0, 5, 20):
+        assert len(truncate_text("x" * 100, limit)) <= limit
+    assert len(truncate_text("x" * 100, 20, parse_mode="MarkdownV2")) <= 20
+
+
+@pytest.mark.parametrize("limit", [4, 15])
+def test_truncate_markdownv2_tiny_limits_never_dangle_an_escape(limit: int) -> None:
+    # Clamping the escaped marker mid-escape would leave a trailing backslash
+    # (a dangling MarkdownV2 \X pair 400s the send); the clamp must back off
+    # the odd backslash run, stay within the cap, and keep the text budget.
+    out = truncate_text("x" * 100, limit, parse_mode="MarkdownV2")
+    assert len(out) <= limit
+    assert _trailing_backslashes(out) % 2 == 0  # no dangling escape at the end
diff --git a/tests/test_sender_retry.py b/tests/test_sender_retry.py
index 120a412..f3fcb98 100644
--- a/tests/test_sender_retry.py
+++ b/tests/test_sender_retry.py
@@ -105,6 +105,7 @@ def test_429_honors_retry_after_without_consuming_budget() -> None:
     outcome = _sender(sleep, max_retries=1).send_with_retry("hi")
     assert outcome.delivered is True
     assert outcome.retries == 0  # 429s do not consume the retry budget
+    assert outcome.rate_limited == 2  # both Retry-After waits still counted
     assert sleep.calls == [7.0, 3.0]  # slept exactly the Retry-After values
 
 
@@ -197,6 +198,7 @@ def test_429_waits_are_capped_so_a_stuck_429_cannot_spin_forever() -> None:
     outcome = _sender(sleep).send_with_retry("hi")
     assert outcome.delivered is False  # gave up after the cap
     assert outcome.retries == 0  # 429s never consume the retry budget
+    assert outcome.rate_limited == _MAX_RATE_LIMIT_WAITS  # every wait counted
     assert len(sleep.calls) == _MAX_RATE_LIMIT_WAITS  # slept exactly cap times
     assert sleep.calls == [1.0] * _MAX_RATE_LIMIT_WAITS
 
@@ -354,3 +356,38 @@ def test_heads_up_notice_failure_is_swallowed() -> None:
     outcome = _sender(sleep).send_with_retry("real log line")
     assert outcome.delivered is True
     assert route.call_count == 3  # send, failed notice, successful resend
+
+
+@respx.mock
+def test_3xx_is_a_permanent_failure_not_success() -> None:
+    # Regression: a 3xx (e.g. a proxy redirect) used to fall through every
+    # branch and return delivered=True while nothing was sent. It must count
+    # as failed and consume no retries.
+    route = respx.post(SEND_URL).mock(return_value=httpx.Response(302))
+    sleep = _RecordingSleep()
+    outcome = _sender(sleep).send_with_retry("hi")
+    assert outcome.delivered is False
+    assert outcome.retries == 0
+    assert route.call_count == 1
+    assert sleep.calls == []
+
+
+@respx.mock
+def test_redirect_with_location_is_never_followed() -> None:
+    # Redirects are disabled on the client: a 302 carrying a Location must
+    # surface as-is and fail — the redirect target must never receive the
+    # sendMessage POST, and its (hypothetical 200) response must never be
+    # mistaken for a delivery.
+    redirect_target = respx.get("https://evil.example/hijack").mock(
+        return_value=httpx.Response(200, json={"ok": True, "result": {"message_id": 9}})
+    )
+    route = respx.post(SEND_URL).mock(
+        return_value=httpx.Response(302, headers={"location": "https://evil.example/hijack"})
+    )
+    sleep = _RecordingSleep()
+    outcome = _sender(sleep).send_with_retry("hi")
+    assert outcome.delivered is False
+    assert outcome.retries == 0
+    assert route.call_count == 1  # one POST, never re-issued against the target
+    assert redirect_target.call_count == 0  # the Location was never fetched
+    assert sleep.calls == []
diff --git a/tests/test_smoke.py b/tests/test_smoke.py
index e29e682..f358916 100644
--- a/tests/test_smoke.py
+++ b/tests/test_smoke.py
@@ -16,6 +16,8 @@
 
 import logging
 import os
+import time
+from collections.abc import Callable
 from typing import Any
 
 import pytest
@@ -167,6 +169,31 @@ def _live_handler(**kwargs: Any) -> TelegramLoggingHandler:
     return TelegramLoggingHandler(**kwargs)
 
 
+def _wait_for_stats(
+    handler: TelegramLoggingHandler,
+    predicate: Callable[[HandlerStats], bool],
+    timeout: float = 30.0,
+) -> HandlerStats:
+    """Poll ``handler.stats`` until ``predicate`` holds; fail on timeout.
+
+    ``close()`` returns after ``shutdown_timeout`` even when the worker is
+    still mid-send (a daemon thread finishing its last batches in the
+    background), so a stats snapshot taken right after close can be
+    mid-flight. Live tests must not race the worker: poll until the expected
+    counters converge, then assert. If the deadline passes without the
+    predicate becoming true, raise instead of returning stale stats so a
+    test can never pass vacuously on a timeout.
+    """
+    deadline = time.monotonic() + timeout
+    while True:
+        stats = handler.stats
+        if predicate(stats):
+            return stats
+        if time.monotonic() >= deadline:
+            raise TimeoutError(f"stats did not converge within {timeout}s: {stats}")
+        time.sleep(0.05)
+
+
 @requires_live
 def test_live_validate_and_deliver_batched_logs() -> None:
     # validate=True does a real getMe at construction; then a batch of records
@@ -187,9 +214,9 @@ def test_live_validate_and_deliver_batched_logs() -> None:
         logger.removeHandler(handler)
         handler.close()  # drains + joins the worker, so all sends complete
 
-    stats = handler.stats
+    stats = _wait_for_stats(handler, lambda s: s.sent == 6)
     assert stats.queued >= 6
-    assert stats.sent >= 1
+    assert stats.sent == 6
     assert stats.failed == 0
 
 
@@ -204,11 +231,12 @@ def test_live_parse_mode_escaping_is_accepted(parse_mode: str) -> None:
     logger.propagate = False
     logger.addHandler(handler)
     try:
-        logger.warning("specials _ * [ ] ( ) ~ ` > # + - = | { } . ! < & > %s", "<b>x</b>")
+        logger.warning("specials _ * [ ] ( ) ~ ` > # + - = | { } . ! < > & %s", "<b>x</b>")
     finally:
         logger.removeHandler(handler)
         handler.close()
-    assert handler.stats.failed == 0
+    stats = _wait_for_stats(handler, lambda s: s.sent >= 1)
+    assert stats.failed == 0
 
 
 @requires_live
@@ -225,5 +253,32 @@ def test_live_split_overflow_delivers_oversized_message() -> None:
     finally:
         logger.removeHandler(handler)
         handler.close()
-    assert handler.stats.sent >= 1
-    assert handler.stats.failed == 0
+    stats = _wait_for_stats(handler, lambda s: s.sent >= 1)
+    assert stats.sent >= 1
+    assert stats.failed == 0
+
+
+@requires_live
+def test_live_batch_size_groups_records_5_and_5() -> None:
+    # 10 real error records (each with a traceback) with batch_size=5 must go
+    # out as exactly 2 messages (5+5), never as singles or one 10-record
+    # message: sent counts records, batches_sent counts messages.
+    handler = _live_handler(batch_size=5, flush_interval=0.5, validate=False)
+    logger = logging.getLogger("tg-smoke.live.batch-size")
+    logger.setLevel(logging.ERROR)
+    logger.propagate = False
+    logger.addHandler(handler)
+    try:
+        for i in range(10):
+            try:
+                raise ValueError(f"intentional smoke error {i + 1}/10")
+            except ValueError:
+                logger.exception("live smoke batch error %d/10", i + 1)
+    finally:
+        logger.removeHandler(handler)
+        handler.close()
+    stats = _wait_for_stats(handler, lambda s: s.sent == 10)
+    assert stats.queued == 10
+    assert stats.sent == 10
+    assert stats.batches_sent == 2
+    assert stats.failed == 0
diff --git a/tests/test_worker.py b/tests/test_worker.py
index 4f52824..5f9b56c 100644
--- a/tests/test_worker.py
+++ b/tests/test_worker.py
@@ -110,3 +110,84 @@ def test_worker_stops_after_flushing_partial_batch_on_shutdown() -> None:
     assert not worker.is_alive()
     assert sender.sent == ["one"]
     assert stats.snapshot().sent == 1
+
+
+def test_worker_stops_when_sentinel_is_lost_to_full_queue() -> None:
+    # Regression: with a saturated queue the handler's sentinel put is
+    # suppressed, and drop_oldest can even evict it. The shutdown event alone
+    # must still stop the worker after everything queued is drained.
+    q: queue.Queue[object] = queue.Queue(maxsize=1)
+    q.put_nowait(_record("only"))
+    sender = _FakeSender()
+    stats = StatsCollector()
+    worker = WorkerThread(
+        q,
+        cast(TelegramSender, sender),
+        stats,
+        lambda r: r.getMessage(),
+        batch_size=1,
+        flush_interval=0.01,
+    )
+    worker.start()
+    worker.shutdown()  # no sentinel involved: the event must be enough
+    worker.join(timeout=3.0)
+    assert not worker.is_alive()
+    assert sender.sent == ["only"]  # drained before exit
+
+
+def test_shutdown_event_interrupts_long_interval_wait() -> None:
+    # Regression: with a partial batch pending inside a long flush_interval,
+    # shutdown used to wait the whole interval out (the mid-batch get only
+    # saw the sentinel, which a saturated queue can lose). The event is now
+    # polled mid-batch, so the worker exits promptly and still flushes the
+    # pending record.
+    q: queue.Queue[object] = queue.Queue()
+    q.put_nowait(_record("pending"))
+    sender = _FakeSender()
+    stats = StatsCollector()
+    worker = WorkerThread(
+        q,
+        cast(TelegramSender, sender),
+        stats,
+        lambda r: r.getMessage(),
+        batch_size=10,  # one record never fills the batch
+        flush_interval=60.0,  # the long wait shutdown must cut short
+    )
+    worker.start()
+    worker.shutdown()  # event only; no sentinel on the queue at all
+    worker.join(timeout=3.0)
+    assert not worker.is_alive()
+    assert sender.sent == ["pending"]  # partial batch flushed before exit
+
+
+def test_worker_counts_rate_limited_waits_in_stats() -> None:
+    class _RateLimitedSender(_FakeSender):
+        def send_with_retry(self, text: str) -> SendOutcome:
+            self.sent.append(text)
+            # 429s never consume the retry budget, but every wait must count.
+            return SendOutcome(delivered=True, retries=0, rate_limited=3)
+
+    stats = _run_worker(_RateLimitedSender(), lambda r: r.getMessage(), [_record("x")])
+    assert stats.snapshot().rate_limited == 3
+
+
+def test_sentinel_consumption_has_matching_task_done() -> None:
+    # Regression: the sentinel is consumed with a get() but never got a
+    # task_done(), so unfinished_tasks stayed > 0 forever and any
+    # queue.join() would hang.
+    q: queue.Queue[object] = queue.Queue()
+    stats = StatsCollector()
+    worker = WorkerThread(
+        q,
+        cast(TelegramSender, _FakeSender()),
+        stats,
+        lambda r: r.getMessage(),
+        batch_size=1,
+        flush_interval=0.01,
+    )
+    worker.start()
+    q.put(_record("x"))
+    q.put(SHUTDOWN)
+    worker.join(timeout=3.0)
+    assert not worker.is_alive()
+    assert q.unfinished_tasks == 0
diff --git a/tg_logging_handler/batching.py b/tg_logging_handler/batching.py
index 06395f1..cc51171 100644
--- a/tg_logging_handler/batching.py
+++ b/tg_logging_handler/batching.py
@@ -20,6 +20,13 @@
 
 __all__ = ["BatchAccumulator"]
 
+# Longest single wait inside a partial batch. Waiting beyond this (a long
+# ``flush_interval``) would keep the worker blind to shutdown for the whole
+# interval; the loop re-checks ``stop`` after every capped wait, so shutdown
+# latency stays bounded to roughly this value even when the SHUTDOWN sentinel
+# is lost to a saturated queue or drop_oldest eviction.
+_MAX_RECORD_WAIT = 1.0
+
 
 class BatchAccumulator:
     """Collect queued records into batches flushed by size or elapsed time.
@@ -51,6 +58,7 @@ def collect(
         record_queue: queue.Queue[object],
         shutdown_sentinel: object,
         timeout: float | None = None,
+        stop: Callable[[], bool] | None = None,
     ) -> tuple[list[LogRecord], bool]:
         """Drain up to one batch from ``record_queue``.
 
@@ -58,6 +66,14 @@ def collect(
         batch is empty. Returns ``([], False)`` if nothing arrived and no
         partial batch was pending.
 
+        ``stop``, when given, is polled after a capped wait that elapsed with
+        nothing arriving; a ``True`` return flushes the partial batch
+        immediately so the caller can honour a shutdown request without
+        waiting out a long ``flush_interval``. Records that keep flowing are
+        never split by ``stop``: once shutdown is requested the caller still
+        wants everything already accepted drained, and a batch still growing
+        would otherwise flush per record.
+
         ``batch`` and ``batch_started`` are locals, not instance state: each
         call drains whatever has arrived and returns it, so a batch never
         spans two calls. (The flush timer starts on the first record and is
@@ -72,6 +88,7 @@ def collect(
             return [], False
         if item is shutdown_sentinel:
             shutdown = True
+            record_queue.task_done()  # the sentinel was consumed like any item
         else:
             assert isinstance(item, LogRecord)  # only LogRecords are enqueued
             batch.append(item)
@@ -83,11 +100,20 @@ def collect(
             if remaining <= 0:
                 break  # interval elapsed → flush partial batch
             try:
-                item = record_queue.get(timeout=remaining)
+                item = record_queue.get(timeout=min(remaining, _MAX_RECORD_WAIT))
             except queue.Empty:
+                if min(remaining, _MAX_RECORD_WAIT) < remaining:
+                    if stop is not None and stop():
+                        # A long capped wait elapsed with nothing arriving and a
+                        # shutdown is pending: flush now. Never checked between
+                        # arriving records — a flowing batch must keep growing.
+                        break
+                    continue  # capped wait elapsed; re-check the interval
                 break  # nothing more arrived within the interval → flush partial
             if item is shutdown_sentinel:
                 shutdown = True
+                # consumed item; batch records are task_done'd by the caller
+                record_queue.task_done()
                 break
             assert isinstance(item, LogRecord)
             batch.append(item)
diff --git a/tg_logging_handler/config.py b/tg_logging_handler/config.py
index bd06b5c..7eb3297 100644
--- a/tg_logging_handler/config.py
+++ b/tg_logging_handler/config.py
@@ -43,12 +43,14 @@ def resolve_credentials(
         raise TelegramConfigError(_invalid_token_message(token))
 
     resolved_chat = chat_id if chat_id is not None else os.environ.get("TG_CHAT_ID")
-    if resolved_chat is None or resolved_chat == "":
+    if resolved_chat is not None:
+        resolved_chat = str(resolved_chat).strip()
+    if not resolved_chat:
         raise TelegramConfigError(
             "No chat_id provided. Pass chat_id=... or set the TG_CHAT_ID environment variable."
         )
 
-    return token, str(resolved_chat)
+    return token, resolved_chat
 
 
 def resolve_topic_id(topic_id: int | None) -> int | None:
@@ -120,7 +122,20 @@ def validate_token(token: str, api_base_url: str) -> None:
             "Pass validate=False to skip this check in offline environments."
         ) from exc
 
-    if response.status_code != 200 or not response.json().get("ok", False):
+    try:
+        body = response.json()
+    except ValueError as exc:
+        # A proxy/gateway error page behind a 200 must not surface as a bare
+        # ValueError; the user is promised TelegramConfigError.
+        raise TelegramConfigError(
+            f"Telegram returned a non-JSON response (HTTP {response.status_code}). "
+            "Check that the bot token is correct."
+        ) from exc
+
+    # The body must be an object with ok == True exactly: a scalar/list body
+    # or a merely truthy ok (1, "true") means the token was not accepted.
+    body_ok = isinstance(body, dict) and body.get("ok") is True
+    if response.status_code != 200 or not body_ok:
         raise TelegramConfigError(
             f"Telegram rejected the token (getMe returned HTTP {response.status_code}). "
             "Check that the bot token is correct."
diff --git a/tg_logging_handler/handler.py b/tg_logging_handler/handler.py
index b41dc9f..112d5a0 100644
--- a/tg_logging_handler/handler.py
+++ b/tg_logging_handler/handler.py
@@ -11,7 +11,8 @@
 import copy
 import logging
 import queue
-from typing import Literal
+import threading
+from typing import Any, Literal, cast
 
 from . import queue_policy
 from .config import resolve_credentials, resolve_topic_id, validate_token
@@ -70,64 +71,102 @@ def __init__(
     ) -> None:
         super().__init__(level=level)
 
-        if batch_size < 1:
-            raise ValueError("batch_size must be >= 1")
-        if flush_interval < 0:
-            raise ValueError("flush_interval must be >= 0")
-        if max_retries < 0:
-            raise ValueError("max_retries must be >= 0")
-        if queue_maxsize < 0:
-            raise ValueError("queue_maxsize must be >= 0")
-
-        resolved_token, resolved_chat = resolve_credentials(token, chat_id)
-        resolved_topic = resolve_topic_id(topic_id)
-        if validate:
-            validate_token(resolved_token, api_base_url)
-
-        # Full signature is live: batching + retry/backoff/429, overflow +
-        # queue policies + full stats, parse_mode escaping. The worker
-        # escapes each record for parse_mode; the sender forwards parse_mode to
-        # sendMessage so Telegram renders the (now-safe) entities.
-        self._shutdown_timeout = shutdown_timeout
-        self._closed = False
-        self._stats = StatsCollector()
-        self._queue: queue.Queue[object] = queue.Queue(maxsize=queue_maxsize)
-        # Validate the policy name now so a typo fails fast at construction, not
-        # silently at the first queue overflow.
-        self._put_policy = queue_policy.select_put_policy(queue_full_policy)
-        self._sender = TelegramSender(
-            resolved_token,
-            resolved_chat,
-            api_base_url,
-            parse_mode=parse_mode,
-            max_retries=max_retries,
-            message_thread_id=resolved_topic,
-        )
-        self._worker = WorkerThread(
-            record_queue=self._queue,
-            sender=self._sender,
-            stats=self._stats,
-            format_record=self.format,
-            batch_size=batch_size,
-            flush_interval=flush_interval,
-            overflow=overflow,
-            parse_mode=parse_mode,
-        )
-        self._worker.start()
-        atexit.register(self.close)
+        try:
+            if batch_size < 1:
+                raise ValueError("batch_size must be >= 1")
+            if flush_interval < 0:
+                raise ValueError("flush_interval must be >= 0")
+            if max_retries < 0:
+                raise ValueError("max_retries must be >= 0")
+            if queue_maxsize < 0:
+                raise ValueError("queue_maxsize must be >= 0")
+
+            resolved_token, resolved_chat = resolve_credentials(token, chat_id)
+            resolved_topic = resolve_topic_id(topic_id)
+            if validate:
+                validate_token(resolved_token, api_base_url)
+
+            # Full signature is live: batching + retry/backoff/429, overflow +
+            # queue policies + full stats, parse_mode escaping. The worker
+            # escapes each record for parse_mode; the sender forwards parse_mode
+            # to sendMessage so Telegram renders the (now-safe) entities.
+            self._shutdown_timeout = shutdown_timeout
+            self._closed = False
+            # Guards the _closed flag against emit(), which checks it and then
+            # enqueues: close() must not interleave its write between the
+            # check and the put, or an in-flight emit would feed a queue no
+            # one drains (block policy: hang forever; otherwise: a record
+            # counted queued but never sent). The worker never takes this
+            # lock, so close() cannot deadlock against the drain.
+            self._close_lock = threading.Lock()
+            self._stats = StatsCollector()
+            self._queue: queue.Queue[object] = queue.Queue(maxsize=queue_maxsize)
+            # Validate the policy name now so a typo fails fast at construction,
+            # not silently at the first queue overflow.
+            self._put_policy = queue_policy.select_put_policy(queue_full_policy)
+            self._sender = TelegramSender(
+                resolved_token,
+                resolved_chat,
+                api_base_url,
+                parse_mode=parse_mode,
+                max_retries=max_retries,
+                message_thread_id=resolved_topic,
+            )
+            self._worker = WorkerThread(
+                record_queue=self._queue,
+                sender=self._sender,
+                stats=self._stats,
+                format_record=self.format,
+                batch_size=batch_size,
+                flush_interval=flush_interval,
+                overflow=overflow,
+                parse_mode=parse_mode,
+            )
+            self._worker.start()
+            atexit.register(self.close)
+        except Exception:
+            # super().__init__ just registered a weakref to this half-built
+            # instance in logging._handlerList; exit-time logging.shutdown()
+            # would call close() on it and print a stray AttributeError after
+            # the real construction error. Drop the weakref, then re-raise.
+            # The cleanup itself may also fail (e.g. logging internals torn
+            # down at interpreter shutdown); that must never mask the real
+            # construction error, so swallow any failure here.
+            with contextlib.suppress(Exception):
+                handler_list = cast(Any, logging)._handlerList
+                with cast(Any, logging)._lock:
+                    handler_list[:] = [ref for ref in handler_list if ref() is not self]
+            raise
 
     def emit(self, record: logging.LogRecord) -> None:
         """Snapshot ``record`` and enqueue it. Never raises (stdlib contract)."""
+        if self._closed:
+            # Fast path; the authoritative check below is under the lock.
+            self._stats.increment("dropped")
+            return
         try:
             item = self._snapshot(record)
-            result = self._put_policy(self._queue, item)
-            # drop_oldest can lose an older record even while enqueueing the new
-            # one, so count both outcomes from the PutResult.
-            if result.enqueued:
-                self._stats.increment("queued")
-            self._stats.increment("dropped", result.dropped)
         except Exception:  # emit must never propagate
             self.handleError(record)
+            return
+        with self._close_lock:
+            if self._closed:
+                # close() interleaved between snapshot and put: no worker
+                # drains this queue any more, so drop and count instead of
+                # enqueueing into a dead pipeline (block policy would hang
+                # this thread forever).
+                self._stats.increment("dropped")
+                return
+            try:
+                result = self._put_policy(self._queue, item)
+            except Exception:  # emit must never propagate
+                self.handleError(record)
+                return
+        # drop_oldest can lose an older record even while enqueueing the new
+        # one, so count both outcomes from the PutResult.
+        if result.enqueued:
+            self._stats.increment("queued")
+        self._stats.increment("dropped", result.dropped)
 
     @staticmethod
     def _snapshot(record: logging.LogRecord) -> logging.LogRecord:
@@ -150,13 +189,22 @@ def stats(self) -> HandlerStats:
 
     def close(self) -> None:
         """Drain the queue, stop the worker, release the client. Idempotent."""
-        if self._closed:
-            return
-        self._closed = True
-        # Queue saturated -> worker still stops via its own drain; worst case we
-        # wait out shutdown_timeout below. ponytail: rare edge, not worth more.
-        with contextlib.suppress(queue.Full):
-            self._queue.put_nowait(SHUTDOWN)
+        with self._close_lock:
+            if self._closed:
+                return
+            self._closed = True
+            # Signal stop twice: the event is authoritative (the sentinel below
+            # is only a wakeup and can be lost to a saturated queue or
+            # drop_oldest eviction); the worker drains everything queued first,
+            # then exits. The sentinel goes in BEFORE the event: the worker's
+            # exit check is "event set AND queue empty", so signaling first
+            # would let it exit while the sentinel is still about to be
+            # enqueued, stranding it in a queue nobody drains (queue.join()
+            # would hang forever). The join below runs outside the lock so an
+            # emit blocked on a full queue (block policy) can finish draining.
+            with contextlib.suppress(queue.Full):
+                self._queue.put_nowait(SHUTDOWN)
+            self._worker.shutdown()
         self._worker.join(timeout=self._shutdown_timeout)
         atexit.unregister(self.close)
         super().close()
diff --git a/tg_logging_handler/overflow.py b/tg_logging_handler/overflow.py
index 89a1da0..9dbd086 100644
--- a/tg_logging_handler/overflow.py
+++ b/tg_logging_handler/overflow.py
@@ -77,6 +77,13 @@ def truncate_text(text: str, max_length: int, parse_mode: str | None = None) ->
     if len(text) <= max_length:
         return text
     marker = escape(_TRUNCATION_MARKER, parse_mode)
+    # A pathological max_length smaller than the marker itself must still
+    # yield a result within the cap (Telegram would 400 an oversized body);
+    # clamp the marker, then keep the remainder of the budget for text. The
+    # clamped marker is nudged off a trailing backslash so a MarkdownV2 \X
+    # pair is never split (a dangling backslash 400s the send).
+    marker = marker[:max_length]
+    marker = marker[: _safe_cut(marker, len(marker))]
     keep = _safe_cut(text, max(max_length - len(marker), 0))
     return text[:keep] + marker
 
diff --git a/tg_logging_handler/queue_policy.py b/tg_logging_handler/queue_policy.py
index e157ccb..2488a01 100644
--- a/tg_logging_handler/queue_policy.py
+++ b/tg_logging_handler/queue_policy.py
@@ -76,6 +76,7 @@ def put_drop_oldest(q: queue.Queue[object], item: object) -> PutResult:
             q.get_nowait()
         except queue.Empty:
             continue  # someone drained it; retry the put
+        q.task_done()  # every get must be matched or queue.join() would hang
         dropped += 1
     if _try_put(q, item):
         return PutResult(enqueued=True, dropped=dropped)
diff --git a/tg_logging_handler/sender.py b/tg_logging_handler/sender.py
index 5c48b5d..404d963 100644
--- a/tg_logging_handler/sender.py
+++ b/tg_logging_handler/sender.py
@@ -38,10 +38,15 @@ class SendOutcome:
     exhausted, after a permanent failure, or after too many consecutive 429
     waits even when the retry budget remains; the worker counts it as
     ``failed`` so ``sent`` stays accurate.
+
+    ``retries`` counts transient backoff retries (429s never consume the retry
+    budget); ``rate_limited`` counts the 429 ``Retry-After`` waits so a
+    rate-limit storm stays visible in the stats even at zero retries.
     """
 
     delivered: bool
     retries: int
+    rate_limited: int = 0
 
 
 class TelegramSender:
@@ -78,6 +83,8 @@ def _get_client(self) -> httpx.Client:
         """Return the worker-thread-local client, creating it on first use."""
         if self._client is None:
             timeout = httpx.Timeout(DEFAULT_READ_TIMEOUT, connect=DEFAULT_CONNECT_TIMEOUT)
+            # Redirects stay disabled (httpx default): a 302/308 with a
+            # Location must surface as-is and fail, never be followed.
             self._client = httpx.Client(timeout=timeout)
         return self._client
 
@@ -96,11 +103,14 @@ def send_with_retry(self, text: str) -> SendOutcome:
         """
         attempts = 0
         rate_limited = 0
+        rate_limited_waits = 0
         delay = _BASE_DELAY_SECONDS
         while True:
             try:
                 self._send_once(text)
-                return SendOutcome(delivered=True, retries=attempts)
+                return SendOutcome(
+                    delivered=True, retries=attempts, rate_limited=rate_limited_waits
+                )
             except TelegramSendError as exc:
                 if exc.retry_after is not None:
                     # 429 rate limit: honor Retry-After without consuming the
@@ -115,15 +125,22 @@ def send_with_retry(self, text: str) -> SendOutcome:
                         self._notified_rate_limit = True
                         self._notify_rate_limited()
                     if rate_limited >= _MAX_RATE_LIMIT_WAITS:
-                        return SendOutcome(delivered=False, retries=attempts)
+                        return SendOutcome(
+                            delivered=False,
+                            retries=attempts,
+                            rate_limited=rate_limited_waits,
+                        )
                     rate_limited += 1
+                    rate_limited_waits += 1
                     # Clamp the server's Retry-After to [0, _MAX_DELAY_SECONDS]
                     # so one huge (or negative) header value can't stall the
                     # worker for hours or crash time.sleep.
                     self._sleep(min(max(exc.retry_after, 0.0), _MAX_DELAY_SECONDS))
                     continue
                 if not exc.retryable or attempts >= self._max_retries:
-                    return SendOutcome(delivered=False, retries=attempts)
+                    return SendOutcome(
+                        delivered=False, retries=attempts, rate_limited=rate_limited_waits
+                    )
                 attempts += 1
                 rate_limited = 0  # a non-429 retry breaks the consecutive-429 run
                 jittered = delay * (1.0 + _JITTER_FRACTION * (2.0 * random.random() - 1.0))
@@ -148,7 +165,10 @@ def _notify_rate_limited(self) -> None:
             payload["message_thread_id"] = self._message_thread_id
         try:
             self._get_client().post(self._url, json=payload)
-        except httpx.HTTPError as exc:
+        except Exception as exc:
+            # Broad by design: any failure in the notice path is noise
+            # (serialization, transport, the very rate limit we report); it
+            # must never surface through send_with_retry or fail a batch.
             _diagnostics.report(f"could not send 429 heads-up notification: {exc}")
 
     def _send_once(self, text: str) -> None:
@@ -194,6 +214,13 @@ def _send_once(self, text: str) -> None:
             raise TelegramSendError(
                 f"server error {response.status_code}: {response.text}", retryable=True
             )
+        if response.status_code != 200:
+            # Anything else — a 3xx (redirects are never followed), or a
+            # non-200 2xx — means nothing was delivered; never report success.
+            raise TelegramSendError(
+                f"unexpected status {response.status_code}: {response.text}",
+                retryable=False,
+            )
 
     def close(self) -> None:
         """Close the underlying client. Safe to call when never opened."""
diff --git a/tg_logging_handler/stats.py b/tg_logging_handler/stats.py
index 5b99a7c..99096c1 100644
--- a/tg_logging_handler/stats.py
+++ b/tg_logging_handler/stats.py
@@ -20,6 +20,7 @@ class HandlerStats:
     sent: int = 0
     batches_sent: int = 0
     retries: int = 0
+    rate_limited: int = 0
     failed: int = 0
     dropped: int = 0
 
diff --git a/tg_logging_handler/worker.py b/tg_logging_handler/worker.py
index ee311f8..19e14fb 100644
--- a/tg_logging_handler/worker.py
+++ b/tg_logging_handler/worker.py
@@ -52,11 +52,19 @@ def __init__(
         self._sender = sender
         self._stats = stats
         self._format_record = format_record
+        # Authoritative stop signal: close() sets it. The SHUTDOWN sentinel is
+        # only a wakeup hint and can be lost (suppressed put on a saturated
+        # queue, drop_oldest eviction), so the loop must not rely on it alone.
+        self._shutdown_event = threading.Event()
         # batch_size==1 (default) makes the accumulator flush every record.
         self._accumulator = BatchAccumulator(batch_size=batch_size, flush_interval=flush_interval)
         self._overflow = overflow
         self._parse_mode = parse_mode
 
+    def shutdown(self) -> None:
+        """Request a stop after everything currently queued has been drained."""
+        self._shutdown_event.set()
+
     def run(self) -> None:
         try:
             self._loop()
@@ -67,8 +75,8 @@ def _loop(self) -> None:
         """Run the drain/send cycle until shutdown.
 
         Failed-count accounting is deliberate:
-        - an exception while processing a batch increments ``failed`` once
-          (the batch, not each record in it);
+        - an exception while processing a batch increments ``failed`` by
+          ``len(batch)`` (every record in that batch never reached the wire);
         - a partial or failed send — ``SendOutcome(delivered=False)`` from
           retries exhausted, a permanent failure, or too many consecutive 429
           waits (``_MAX_RATE_LIMIT_WAITS``) even when the retry budget
@@ -80,18 +88,28 @@ def _loop(self) -> None:
         delivered split still flags the batch).
         """
         while True:
+            if self._shutdown_event.is_set() and self._queue.empty():
+                # The sentinel can be lost (suppressed put on a saturated
+                # queue, drop_oldest eviction); the event is the reliable stop
+                # signal. Everything queued before shutdown is drained first.
+                return
             # Empty batch -> block up to 1s waiting for a record (a quiet
             # handler must not busy-poll); partial batch -> collect until size
-            # or interval, whichever comes first. Shutdown flushes any
-            # partial batch first, then stops.
-            batch, shutdown = self._accumulator.collect(self._queue, SHUTDOWN, timeout=1.0)
+            # or interval, whichever comes first. The accumulator polls the
+            # shutdown event between its capped waits, so a shutdown is
+            # honoured within ~1s even when a long flush_interval has a
+            # partial batch pending and the sentinel was lost. Every batch is
+            # flushed first; nothing already accepted is dropped.
+            batch, shutdown = self._accumulator.collect(
+                self._queue, SHUTDOWN, timeout=1.0, stop=self._shutdown_event.is_set
+            )
             if batch:
                 try:
                     self._process_batch(batch)
                 except Exception as exc:  # broad by design, see comment below
                     # A single bad record (format error) must never take the
                     # worker thread down.
-                    self._stats.increment("failed")
+                    self._stats.increment("failed", len(batch))
                     _diagnostics.report(f"failed to send batch: {exc}")
                 finally:
                     for _ in batch:
@@ -141,6 +159,9 @@ def _process_batch(self, batch: list[LogRecord]) -> None:
         for message in messages:
             outcome = self._sender.send_with_retry(message)
             self._stats.increment("retries", outcome.retries)
+            # 429 Retry-After waits count separately: they never consume the
+            # retry budget, but a rate-limit storm must still be visible.
+            self._stats.increment("rate_limited", outcome.rate_limited)
             delivered_all = delivered_all and outcome.delivered
 
         if delivered_all:
diff --git a/uv.lock b/uv.lock
index 88c45fc..4275428 100644
--- a/uv.lock
+++ b/uv.lock
@@ -512,7 +512,7 @@ wheels = [
 
 [[package]]
 name = "tg-logging-handler"
-version = "0.1.3"
+version = "0.1.4"
 source = { editable = "." }
 dependencies = [
     { name = "httpx" },