diff --git a/docs/02_concepts/13_http_compression.mdx b/docs/02_concepts/13_http_compression.mdx
index f5fa9609..b7f39733 100644
--- a/docs/02_concepts/13_http_compression.mdx
+++ b/docs/02_concepts/13_http_compression.mdx
@@ -15,7 +15,11 @@ The Apify client compresses request bodies before sending them to the API. It re
 
 ## How it works
 
-The client compresses request bodies using the compressor configured via the `compression` parameter (default `'gzip'`). The server supports both gzip and brotli and decompresses the request body transparently.
+The client compresses request bodies using the compressor configured via the `compression` parameter (default `'gzip'`). The server supports both gzip and brotli and decompresses the request body transparently. A body is compressed only when it is large enough to benefit and its content type isn't already compressed, as the next two sections describe.
+
+## Minimum body size
+
+The client sends bodies smaller than 1024 bytes without compression and without the `Content-Encoding` header. A body of this size fits in one network packet, so compression doesn't remove a network round trip and only costs CPU time. For very small bodies, the compression format adds bytes and can make the body larger.
 
 ## Already-compressed payloads
 
@@ -84,7 +88,7 @@ client = ApifyClient(token='MY-APIFY-TOKEN', compression=BrotliHttpCompressor(qu
 client = ApifyClient(token='MY-APIFY-TOKEN', compression=GzipHttpCompressor(quality=9))
 ```
 
-You can also implement a fully custom compressor by subclassing `HttpCompressor`:
+You can also implement a fully custom compressor by subclassing `HttpCompressor`. The client calls it only for bodies that reach the [minimum body size](#minimum-body-size) and aren't [already compressed](#already-compressed-payloads):
 
 ```python
 from apify_client import ApifyClient
diff --git a/src/apify_client/_consts.py b/src/apify_client/_consts.py
index b41bc3a4..d64fa1ef 100644
--- a/src/apify_client/_consts.py
+++ b/src/apify_client/_consts.py
@@ -35,6 +35,13 @@
 OVERRIDABLE_DEFAULT_HEADERS = {'Accept', 'Authorization', 'Accept-Encoding', 'User-Agent'}
 """Headers that can be overridden by users, but will trigger a warning if they do so, as it may lead to API errors."""
 
+MIN_COMPRESSION_SIZE = 1024
+"""Smallest request body, in bytes, that is worth compressing.
+
+A smaller body already fits in a single network packet, so compressing it costs CPU time without
+saving a round trip.
+"""
+
 ALREADY_COMPRESSED_MEDIA_TYPE_PREFIXES = ('audio/', 'image/', 'video/')
 """Media type prefixes whose payloads carry their own compression, so compressing the request body is wasted work."""
 
diff --git a/src/apify_client/http_clients/_base.py b/src/apify_client/http_clients/_base.py
index 3862aa68..6b9fc6d6 100644
--- a/src/apify_client/http_clients/_base.py
+++ b/src/apify_client/http_clients/_base.py
@@ -16,6 +16,7 @@
     DEFAULT_TIMEOUT_MAX,
     DEFAULT_TIMEOUT_MEDIUM,
     DEFAULT_TIMEOUT_SHORT,
+    MIN_COMPRESSION_SIZE,
 )
 from apify_client._docs import docs_group
 from apify_client._statistics import ClientStatistics
@@ -223,6 +224,24 @@ def _compute_timeout(self, timeout: Timeout, *, attempt: int) -> int | float | N
         new_timeout = min(resolved * (2 ** (attempt - 1)), self._timeout_max)
         return to_seconds(new_timeout)
 
+    @staticmethod
+    def _is_body_worth_compressing(data: str | bytes | bytearray | None) -> bool:
+        """Whether this body clears the size threshold `_prepare_request_call` compresses at, cheaply.
+
+        Below the threshold nothing is ever compressed. At or above it the content type still decides, but
+        checking that here would buy nothing - a body that turns out to be already compressed only wastes the
+        thread hop this answer guards.
+
+        The threshold is measured on encoded bytes, so a character count alone cannot decide a `str`. It is a
+        lower bound, so a `str` long enough in characters is long enough in bytes too. Below that the encoded
+        length decides, and the body is then under 4 KiB, so encoding it here is cheap.
+        """
+        if isinstance(data, str):
+            return len(data) >= MIN_COMPRESSION_SIZE or len(data.encode('utf-8')) >= MIN_COMPRESSION_SIZE
+        if isinstance(data, (bytes, bytearray)):
+            return len(data) >= MIN_COMPRESSION_SIZE
+        return False
+
     def _prepare_request_call(
         self,
         *,
@@ -234,10 +253,11 @@ def _prepare_request_call(
         """Prepare headers, params, and body for an HTTP request.
 
         Merges the client's default headers (including authorization) with per-request headers, serializes JSON
-        and compresses the body unless its content type says the payload is already compressed. Header names are
-        treated case-insensitively and per-request values win over the client defaults. For JSON bodies, a
-        `Content-Type` header is set unless the caller supplied one. `Content-Encoding` always describes what was
-        actually applied to the body, so a caller-supplied value is dropped whenever nothing was compressed.
+        and compresses the body unless it is smaller than `MIN_COMPRESSION_SIZE` or its content type says the
+        payload is already compressed. Header names are treated case-insensitively and per-request values win
+        over the client defaults. For JSON bodies, a `Content-Type` header is set unless the caller supplied one.
+        `Content-Encoding` always describes what was actually applied to the body, so a caller-supplied value is
+        dropped whenever nothing was compressed.
         """
         if json is not None and data is not None:
             raise ValueError('Cannot pass both "json" and "data" parameters at the same time!')
@@ -259,7 +279,7 @@ def _prepare_request_call(
                 data = bytes(data)
 
             content_type = next((value for key, value in headers.items() if key.lower() == 'content-type'), None)
-            if is_compressible_content_type(content_type):
+            if len(data) >= MIN_COMPRESSION_SIZE and is_compressible_content_type(content_type):
                 data = self._http_compressor.compress(data)
                 headers = self._merge_headers(headers, {'Content-Encoding': self._http_compressor.content_encoding})
                 compressed = True
diff --git a/src/apify_client/http_clients/_impit.py b/src/apify_client/http_clients/_impit.py
index a082e962..da7a013b 100644
--- a/src/apify_client/http_clients/_impit.py
+++ b/src/apify_client/http_clients/_impit.py
@@ -395,9 +395,10 @@ async def call(
         self._statistics.calls += 1
 
         # Serializing and compressing a request body is CPU-bound and would block the event loop, so
-        # offload request preparation to a worker thread whenever there is a body. Bodyless requests
-        # skip the thread hop, as they have no expensive work to move off the loop.
-        if json is not None or data is not None:
+        # offload preparation to a worker thread whenever there is something to compress. A body the
+        # client sends as it is costs less to prepare inline than the hop itself. A `json` body always
+        # hops, as its size is only known once serialized.
+        if json is not None or self._is_body_worth_compressing(data):
             prepared_headers, prepared_params, content = await asyncio.to_thread(
                 self._prepare_request_call,
                 headers=headers,
diff --git a/tests/unit/test_client_request_queue.py b/tests/unit/test_client_request_queue.py
index bbb2f784..8c34b025 100644
--- a/tests/unit/test_client_request_queue.py
+++ b/tests/unit/test_client_request_queue.py
@@ -120,10 +120,15 @@ def _make_large_requests() -> list[RequestDraftDict]:
 
 
 def _payload_capturing_handler(payloads: list[bytes]) -> Callable[[Request], Response]:
-    """Return a handler that records each POST body (gzip-decompressed) and responds with an empty batch result."""
+    """Return a handler that records each POST body and responds with an empty batch result.
+
+    Bodies below the client's compression threshold arrive uncompressed, so the recorded payload is
+    decompressed only when the request says it was encoded.
+    """
 
     def handler(request: Request) -> Response:
-        payloads.append(gzip.decompress(request.get_data()))
+        body = request.get_data()
+        payloads.append(gzip.decompress(body) if request.headers.get('Content-Encoding') == 'gzip' else body)
         return Response(_EMPTY_BATCH_RESPONSE_CONTENT, status=200, content_type='application/json')
 
     return handler
diff --git a/tests/unit/test_http_clients.py b/tests/unit/test_http_clients.py
index 6755728a..c52e7c2a 100644
--- a/tests/unit/test_http_clients.py
+++ b/tests/unit/test_http_clients.py
@@ -2,10 +2,11 @@
 
 import asyncio
 import gzip
-import io
+import json
 import threading
 import time
 from datetime import UTC, datetime, timedelta
+from io import BytesIO
 from typing import TYPE_CHECKING, cast
 from unittest.mock import AsyncMock, Mock
 
@@ -13,6 +14,7 @@
 import impit
 import pytest
 
+from apify_client._consts import MIN_COMPRESSION_SIZE
 from apify_client._statistics import ClientStatistics
 from apify_client.errors import InvalidResponseBodyError
 from apify_client.http_clients import HttpClient, HttpClientAsync, HttpResponse, ImpitHttpClient, ImpitHttpClientAsync
@@ -25,6 +27,8 @@
     from collections.abc import Callable
     from typing import Any
 
+    from apify_client.types import JsonSerializable
+
 
 class _ConcreteHttpClient(HttpClient):
     """Minimal concrete HttpClient for testing base class helpers."""
@@ -341,125 +345,164 @@ def test_prepare_request_call_basic() -> None:
     assert data is None
 
 
-def test_prepare_request_call_with_json(compressor_case: tuple) -> None:
-    """Test _prepare_request_call with JSON data."""
-    compressor, content_encoding, decompress = compressor_case
-    client = _ConcreteHttpClient(http_compressor=compressor)
+def test_prepare_request_call_with_json() -> None:
+    """A small JSON body is serialized and typed, but sent uncompressed and without a `Content-Encoding`."""
+    client = _ConcreteHttpClient()
 
     json_data = {'key': 'value', 'number': 42}
     headers, _params, data = client._prepare_request_call(json=json_data)
 
     assert headers['Content-Type'] == 'application/json'
-    assert headers['Content-Encoding'] == content_encoding
-    assert data is not None
-    assert isinstance(data, bytes)
-    assert decompress(data) == b'{"key": "value", "number": 42}'
+    assert data == b'{"key": "value", "number": 42}'
+    assert not any(key.lower() == 'content-encoding' for key in headers)
 
 
-def test_prepare_request_call_with_empty_dict_json(compressor_case: tuple) -> None:
-    """Test _prepare_request_call with empty dict JSON (falsy but valid)."""
-    compressor, content_encoding, decompress = compressor_case
-    client = _ConcreteHttpClient(http_compressor=compressor)
+@pytest.mark.parametrize(
+    ('json_body', 'expected'),
+    [
+        pytest.param({}, b'{}', id='empty dict'),
+        pytest.param([], b'[]', id='empty list'),
+        pytest.param(0, b'0', id='zero'),
+        pytest.param(False, b'false', id='false'),
+        pytest.param('', b'""', id='empty string'),
+    ],
+)
+def test_prepare_request_call_with_falsy_json(json_body: JsonSerializable, expected: bytes) -> None:
+    """A falsy but valid JSON body is still serialized and sent, rather than treated as no body at all."""
+    client = _ConcreteHttpClient()
 
-    headers, _params, data = client._prepare_request_call(json={})
+    headers, _params, data = client._prepare_request_call(json=json_body)
 
     assert headers['Content-Type'] == 'application/json'
-    assert headers['Content-Encoding'] == content_encoding
-    assert data is not None
-    assert isinstance(data, bytes)
-    assert decompress(data) == b'{}'
+    assert data == expected
 
 
-def test_prepare_request_call_with_empty_list_json(compressor_case: tuple) -> None:
-    """Test _prepare_request_call with empty list JSON (falsy but valid)."""
-    compressor, content_encoding, decompress = compressor_case
-    client = _ConcreteHttpClient(http_compressor=compressor)
+@pytest.mark.parametrize(
+    'data',
+    [
+        pytest.param('test string', id='str'),
+        pytest.param(b'test bytes', id='bytes'),
+        pytest.param(bytearray(b'test bytearray'), id='bytearray'),
+    ],
+)
+def test_prepare_request_call_with_data(data: str | bytes | bytearray) -> None:
+    """A raw body of any accepted type is normalized to bytes."""
+    client = _ConcreteHttpClient()
 
-    headers, _params, data = client._prepare_request_call(json=[])
+    _headers, _params, prepared = client._prepare_request_call(data=data)
 
-    assert headers['Content-Type'] == 'application/json'
-    assert headers['Content-Encoding'] == content_encoding
-    assert data is not None
-    assert isinstance(data, bytes)
-    assert decompress(data) == b'[]'
+    expected = data.encode('utf-8') if isinstance(data, str) else bytes(data)
+    assert prepared == expected
 
 
-def test_prepare_request_call_with_zero_json(compressor_case: tuple) -> None:
-    """Test _prepare_request_call with zero JSON (falsy but valid)."""
+@pytest.mark.parametrize(
+    'body_size',
+    [
+        pytest.param(MIN_COMPRESSION_SIZE, id='at threshold'),
+        pytest.param(MIN_COMPRESSION_SIZE + 1, id='above threshold'),
+        pytest.param(MIN_COMPRESSION_SIZE * 16, id='well above threshold'),
+    ],
+)
+def test_prepare_request_call_compresses_body_at_or_above_threshold(
+    compressor_case: tuple,
+    body_size: int,
+) -> None:
+    """A raw body of at least `MIN_COMPRESSION_SIZE` bytes is compressed and labeled with its encoding."""
     compressor, content_encoding, decompress = compressor_case
     client = _ConcreteHttpClient(http_compressor=compressor)
+    body = b'x' * body_size
 
-    headers, _params, data = client._prepare_request_call(json=0)
+    headers, _params, data = client._prepare_request_call(data=body)
 
-    assert headers['Content-Type'] == 'application/json'
     assert headers['Content-Encoding'] == content_encoding
-    assert data is not None
-    assert isinstance(data, bytes)
-    assert decompress(data) == b'0'
+    assert decompress(data) == body
 
 
-def test_prepare_request_call_with_false_json(compressor_case: tuple) -> None:
-    """Test _prepare_request_call with False JSON (falsy but valid)."""
-    compressor, content_encoding, decompress = compressor_case
+@pytest.mark.parametrize(
+    'body_size',
+    [
+        pytest.param(0, id='empty'),
+        pytest.param(1, id='single byte'),
+        pytest.param(MIN_COMPRESSION_SIZE - 1, id='just below threshold'),
+    ],
+)
+def test_prepare_request_call_skips_compression_below_threshold(compressor_case: tuple, body_size: int) -> None:
+    """A raw body under `MIN_COMPRESSION_SIZE` is sent verbatim with no `Content-Encoding`, whichever compressor."""
+    compressor, _content_encoding, _decompress = compressor_case
     client = _ConcreteHttpClient(http_compressor=compressor)
+    body = b'x' * body_size
 
-    headers, _params, data = client._prepare_request_call(json=False)
+    headers, _params, data = client._prepare_request_call(data=body)
 
-    assert headers['Content-Type'] == 'application/json'
-    assert headers['Content-Encoding'] == content_encoding
-    assert data is not None
-    assert isinstance(data, bytes)
-    assert decompress(data) == b'false'
+    assert data == body
+    assert not any(key.lower() == 'content-encoding' for key in headers)
 
 
-def test_prepare_request_call_with_empty_string_json(compressor_case: tuple) -> None:
-    """Test _prepare_request_call with empty string JSON (falsy but valid)."""
+def test_prepare_request_call_compresses_bytearray_data(compressor_case: tuple) -> None:
+    """A `bytearray` body above the threshold is compressed without error (regression: needs bytes conversion)."""
     compressor, content_encoding, decompress = compressor_case
     client = _ConcreteHttpClient(http_compressor=compressor)
+    body = bytearray(b'test bytearray' * 128)
 
-    headers, _params, data = client._prepare_request_call(json='')
+    headers, _params, data = client._prepare_request_call(data=body)
 
-    assert headers['Content-Type'] == 'application/json'
     assert headers['Content-Encoding'] == content_encoding
-    assert data is not None
-    assert isinstance(data, bytes)
-    assert decompress(data) == b'""'
+    assert decompress(data) == bytes(body)
 
 
-def test_prepare_request_call_with_string_data(compressor_case: tuple) -> None:
-    """Test _prepare_request_call with string data."""
+def test_prepare_request_call_compresses_json_above_threshold(compressor_case: tuple) -> None:
+    """A JSON body that serializes to at least `MIN_COMPRESSION_SIZE` bytes is compressed."""
     compressor, content_encoding, decompress = compressor_case
     client = _ConcreteHttpClient(http_compressor=compressor)
+    json_data = {'key': 'x' * MIN_COMPRESSION_SIZE}
 
-    headers, _params, data = client._prepare_request_call(data='test string')
+    headers, _params, data = client._prepare_request_call(json=json_data)
 
+    assert headers['Content-Type'] == 'application/json'
     assert headers['Content-Encoding'] == content_encoding
-    assert isinstance(data, bytes)
-    assert decompress(data) == b'test string'
+    assert json.loads(decompress(data)) == json_data
 
 
-def test_prepare_request_call_with_bytes_data(compressor_case: tuple) -> None:
-    """Test _prepare_request_call with bytes data."""
+def test_prepare_request_call_measures_threshold_in_bytes_not_characters(compressor_case: tuple) -> None:
+    """A `str` body under the threshold in characters but over it in UTF-8 bytes is still compressed."""
     compressor, content_encoding, decompress = compressor_case
     client = _ConcreteHttpClient(http_compressor=compressor)
+    # U+00E9 encodes to 2 bytes, so this body is under the threshold in characters but over it in bytes.
+    body = '\u00e9' * (MIN_COMPRESSION_SIZE // 2 + 1)
 
-    headers, _params, data = client._prepare_request_call(data=b'test bytes')
+    headers, _params, data = client._prepare_request_call(data=body)
 
     assert headers['Content-Encoding'] == content_encoding
-    assert isinstance(data, bytes)
-    assert decompress(data) == b'test bytes'
+    assert decompress(data) == body.encode('utf-8')
 
 
-def test_prepare_request_call_with_bytearray_data(compressor_case: tuple) -> None:
-    """Test _prepare_request_call with bytearray data (regression: must compress without error)."""
-    compressor, content_encoding, decompress = compressor_case
-    client = _ConcreteHttpClient(http_compressor=compressor)
+@pytest.mark.parametrize(
+    'data',
+    [
+        pytest.param(b'x' * MIN_COMPRESSION_SIZE, id='bytes at threshold'),
+        pytest.param(bytearray(b'x' * MIN_COMPRESSION_SIZE), id='bytearray at threshold'),
+        pytest.param('x' * MIN_COMPRESSION_SIZE, id='ascii str at threshold'),
+        pytest.param('\u00e9' * (MIN_COMPRESSION_SIZE // 2 + 1), id='multibyte str above byte threshold'),
+    ],
+)
+def test_is_body_worth_compressing(data: Any) -> None:
+    """The gate reports a body `_prepare_request_call` would compress, judging a `str` by its encoded bytes."""
+    assert _ConcreteHttpClient._is_body_worth_compressing(data)
 
-    headers, _params, data = client._prepare_request_call(data=bytearray(b'test bytearray'))
 
-    assert headers['Content-Encoding'] == content_encoding
-    assert isinstance(data, bytes)
-    assert decompress(data) == b'test bytearray'
+@pytest.mark.parametrize(
+    'data',
+    [
+        pytest.param(None, id='no body'),
+        pytest.param(b'x' * (MIN_COMPRESSION_SIZE - 1), id='bytes below threshold'),
+        pytest.param('x' * (MIN_COMPRESSION_SIZE - 1), id='ascii str below threshold'),
+        pytest.param(BytesIO(b'x' * MIN_COMPRESSION_SIZE), id='file-like'),
+        pytest.param({'key': 'x' * MIN_COMPRESSION_SIZE}, id='mapping'),
+    ],
+)
+def test_is_body_not_worth_compressing(data: Any) -> None:
+    """A body below the threshold, or of a type the client sends as it is, needs no worker-thread hop."""
+    assert not _ConcreteHttpClient._is_body_worth_compressing(data)
 
 
 @pytest.mark.parametrize(
@@ -473,13 +516,15 @@ def test_prepare_request_call_with_bytearray_data(compressor_case: tuple) -> Non
 def test_prepare_request_call_skips_compression_for_already_compressed_content(content_type: str) -> None:
     """An already-compressed body is sent verbatim, carries no `Content-Encoding`, and keeps every other header."""
     client = _ConcreteHttpClient(token='test_token', http_compressor=GzipHttpCompressor())
+    # Above the size threshold, so the content type is what skips compression here.
+    payload = b'\x89PNG' + b'\xff' * MIN_COMPRESSION_SIZE
 
     headers, _params, data = client._prepare_request_call(
         headers={'content-type': content_type},
-        data=b'\x89PNG binary',
+        data=payload,
     )
 
-    assert data == b'\x89PNG binary'
+    assert data == payload
     assert not any(key.lower() == 'content-encoding' for key in headers)
     assert headers['Authorization'] == 'Bearer test_token'
     assert headers['content-type'] == content_type
@@ -489,20 +534,22 @@ def test_prepare_request_call_skips_compression_for_already_compressed_content(c
 def test_prepare_request_call_drops_caller_content_encoding_when_compression_is_skipped() -> None:
     """Skipping compression also strips a caller-supplied `Content-Encoding`, which would misdescribe the body."""
     client = _ConcreteHttpClient(http_compressor=GzipHttpCompressor())
+    # Above the size threshold, so the content type is what skips compression here.
+    payload = b'\xff' * MIN_COMPRESSION_SIZE
 
     headers, _params, data = client._prepare_request_call(
         headers={'content-type': 'image/jpeg', 'content-encoding': 'br'},
-        data=b'jpeg binary',
+        data=payload,
     )
 
-    assert data == b'jpeg binary'
+    assert data == payload
     assert not any(key.lower() == 'content-encoding' for key in headers)
 
 
 def test_prepare_request_call_drops_caller_content_encoding_for_a_streamed_body() -> None:
     """A body that is streamed rather than compressed, such as a file-like object, also loses `Content-Encoding`."""
     client = _ConcreteHttpClient(http_compressor=GzipHttpCompressor())
-    stream = io.BytesIO(b'raw payload')
+    stream = BytesIO(b'raw payload')
 
     headers, _params, data = client._prepare_request_call(
         headers={'content-encoding': 'gzip'},
@@ -528,15 +575,16 @@ def test_prepare_request_call_compresses_exceptions_to_compressed_prefixes(
     """Types that are text or raw are compressed even when they sit under an already-compressed prefix."""
     compressor, content_encoding, decompress = compressor_case
     client = _ConcreteHttpClient(http_compressor=compressor)
+    payload = b'x' * MIN_COMPRESSION_SIZE
 
     headers, _params, data = client._prepare_request_call(
         headers={'content-type': content_type},
-        data=b'raw payload',
+        data=payload,
     )
 
     assert headers['Content-Encoding'] == content_encoding
     assert isinstance(data, bytes)
-    assert decompress(data) == b'raw payload'
+    assert decompress(data) == payload
 
 
 def test_prepare_request_call_json_and_data_error() -> None:
@@ -601,12 +649,25 @@ def test_prepare_request_call_replaces_caller_content_encoding() -> None:
     """A compressed body reports the compressor actually applied, replacing any caller-supplied Content-Encoding."""
     client = _ConcreteHttpClient(http_compressor=GzipHttpCompressor())
 
-    headers, _params, _data = client._prepare_request_call(headers={'content-encoding': 'br'}, data='payload')
+    headers, _params, _data = client._prepare_request_call(
+        headers={'content-encoding': 'br'},
+        data='x' * MIN_COMPRESSION_SIZE,
+    )
 
     encoding_headers = {key: value for key, value in headers.items() if key.lower() == 'content-encoding'}
     assert encoding_headers == {'Content-Encoding': 'gzip'}
 
 
+def test_prepare_request_call_drops_caller_content_encoding_when_skipping_compression() -> None:
+    """A caller-supplied Content-Encoding is dropped for an uncompressed body, so it cannot mislabel it."""
+    client = _ConcreteHttpClient(http_compressor=GzipHttpCompressor())
+
+    headers, _params, data = client._prepare_request_call(headers={'content-encoding': 'br'}, data='payload')
+
+    assert data == b'payload'
+    assert not any(key.lower() == 'content-encoding' for key in headers)
+
+
 def test_build_url_with_params_none() -> None:
     """Test _build_url_with_params with None params."""
     client = _ConcreteHttpClient()
@@ -667,27 +728,85 @@ async def test_async_call_compresses_request_body_off_the_event_loop() -> None:
     client = ImpitHttpClientAsync(token='test_token', http_compressor=compressor)
     client._impit_async_client = Mock(request=AsyncMock(return_value=Mock(status_code=200)))
 
-    await client.call(method='POST', url='https://api.test.com/endpoint', json={'key': 'value'})
+    await client.call(
+        method='POST',
+        url='https://api.test.com/endpoint',
+        json={'key': 'x' * MIN_COMPRESSION_SIZE},
+    )
 
     assert compressor.compress_thread_id is not None
     assert compressor.compress_thread_id != threading.get_ident()
 
 
+async def test_async_call_compresses_a_multibyte_str_body_off_the_event_loop() -> None:
+    """A `str` body over the threshold only once encoded is still compressed, so it must be offloaded."""
+    compressor = _ThreadRecordingCompressor()
+    client = ImpitHttpClientAsync(token='test_token', http_compressor=compressor)
+    client._impit_async_client = Mock(request=AsyncMock(return_value=Mock(status_code=200)))
+
+    await client.call(
+        method='PUT',
+        url='https://api.test.com/endpoint',
+        data='\u00e9' * (MIN_COMPRESSION_SIZE // 2 + 1),
+    )
+
+    assert compressor.compress_thread_id is not None
+    assert compressor.compress_thread_id != threading.get_ident()
+
+
+def _to_thread_spy(monkeypatch: pytest.MonkeyPatch) -> Mock:
+    """Patch `asyncio.to_thread` with a `Mock` that still performs the hop, to record whether it was used."""
+    spy = Mock(side_effect=asyncio.to_thread)
+    monkeypatch.setattr(asyncio, 'to_thread', spy)
+    return spy
+
+
 async def test_async_call_skips_thread_offload_without_a_body(monkeypatch: pytest.MonkeyPatch) -> None:
     """A bodyless request has nothing to compress, so it must not pay the worker-thread hop."""
     client = ImpitHttpClientAsync(token='test_token')
     client._impit_async_client = Mock(request=AsyncMock(return_value=Mock(status_code=200)))
+    spy = _to_thread_spy(monkeypatch)
 
-    offloaded = False
-    real_to_thread = asyncio.to_thread
+    await client.call(method='GET', url='https://api.test.com/endpoint')
 
-    async def spy_to_thread(func: Any, /, *args: Any, **kwargs: Any) -> Any:
-        nonlocal offloaded
-        offloaded = True
-        return await real_to_thread(func, *args, **kwargs)
+    spy.assert_not_called()
 
-    monkeypatch.setattr(asyncio, 'to_thread', spy_to_thread)
 
-    await client.call(method='GET', url='https://api.test.com/endpoint')
+async def test_async_call_skips_thread_offload_for_a_body_below_the_threshold(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """A raw body too small to be compressed must not pay the worker-thread hop either."""
+    client = ImpitHttpClientAsync(token='test_token')
+    client._impit_async_client = Mock(request=AsyncMock(return_value=Mock(status_code=200)))
+    spy = _to_thread_spy(monkeypatch)
+
+    await client.call(method='PUT', url='https://api.test.com/endpoint', data=b'x' * (MIN_COMPRESSION_SIZE - 1))
+
+    spy.assert_not_called()
+
+
+async def test_async_call_offloads_a_body_at_the_threshold(monkeypatch: pytest.MonkeyPatch) -> None:
+    """A raw body large enough to be compressed is prepared in a worker thread."""
+    client = ImpitHttpClientAsync(token='test_token')
+    client._impit_async_client = Mock(request=AsyncMock(return_value=Mock(status_code=200)))
+    spy = _to_thread_spy(monkeypatch)
+
+    await client.call(method='PUT', url='https://api.test.com/endpoint', data=b'x' * MIN_COMPRESSION_SIZE)
+
+    spy.assert_called_once()
+
+
+async def test_async_call_skips_thread_offload_for_a_body_it_cannot_compress(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """A body of a type the client passes through needs no hop, and deciding that must not need its length."""
+    client = ImpitHttpClientAsync(token='test_token')
+    client._impit_async_client = Mock(request=AsyncMock(return_value=Mock(status_code=200)))
+    spy = _to_thread_spy(monkeypatch)
+
+    # `encode_key_value_store_record_value` passes file-like bodies through, so the gate cannot assume a length.
+    body: Any = BytesIO(b'x' * MIN_COMPRESSION_SIZE)
+
+    await client.call(method='PUT', url='https://api.test.com/endpoint', data=body)
 
-    assert offloaded is False
+    spy.assert_not_called()
diff --git a/tests/unit/test_http_compressors.py b/tests/unit/test_http_compressors.py
index de858f45..d5643ee9 100644
--- a/tests/unit/test_http_compressors.py
+++ b/tests/unit/test_http_compressors.py
@@ -2,18 +2,26 @@
 
 import gzip
 import importlib
+import json
 import sys
 from contextlib import contextmanager
 from typing import TYPE_CHECKING
 
 import brotli
 import pytest
+from werkzeug import Request, Response
 
+from apify_client import ApifyClient, ApifyClientAsync
+from apify_client._consts import MIN_COMPRESSION_SIZE
 from apify_client.http_compressors import BrotliHttpCompressor, GzipHttpCompressor
 from apify_client.http_compressors._resolve import resolve_compressor
 
 if TYPE_CHECKING:
-    from collections.abc import Iterator
+    from collections.abc import Callable, Iterator
+
+    from pytest_httpserver import HTTPServer
+
+    from apify_client.types import HttpCompressionAlgorithm
 
 
 @contextmanager
@@ -147,3 +155,69 @@ def test_resolve_compressor_brotli_raises_clear_error_when_extra_missing() -> No
     """Resolving `compression='brotli'` without the extra raises `ImportError` at resolution time."""
     with _brotli_unavailable(), pytest.raises(ImportError):
         resolve_compressor('brotli')
+
+
+_ITEMS_PATH = '/v2/datasets/test_dataset_id/items'
+_LARGE_BODY = [{'index': index, 'url': f'https://example.com/item/{index}'} for index in range(MIN_COMPRESSION_SIZE)]
+
+
+def _capture_body(captured: list[Request]) -> Callable[[Request], Response]:
+    def handler(request: Request) -> Response:
+        captured.append(request)
+        return Response(status=201, mimetype='application/json')
+
+    return handler
+
+
+@pytest.mark.parametrize(
+    ('compression', 'content_encoding', 'decompress'),
+    [
+        pytest.param('gzip', 'gzip', gzip.decompress, id='gzip'),
+        pytest.param('brotli', 'br', brotli.decompress, id='brotli'),
+    ],
+)
+def test_configured_compression_reaches_the_wire_sync(
+    httpserver: HTTPServer,
+    compression: HttpCompressionAlgorithm,
+    content_encoding: str,
+    decompress: Callable[[bytes], bytes],
+) -> None:
+    """A body above the threshold arrives at the server compressed with the algorithm the client was given."""
+    captured: list[Request] = []
+    httpserver.expect_request(_ITEMS_PATH, method='POST').respond_with_handler(_capture_body(captured))
+
+    api_url = httpserver.url_for('/').removesuffix('/')
+    client = ApifyClient(token='test_token', api_url=api_url, compression=compression)
+
+    client.dataset('test_dataset_id').push_items(_LARGE_BODY)
+
+    assert len(captured) == 1
+    assert captured[0].headers['Content-Encoding'] == content_encoding
+    assert json.loads(decompress(captured[0].get_data())) == _LARGE_BODY
+
+
+@pytest.mark.parametrize(
+    ('compression', 'content_encoding', 'decompress'),
+    [
+        pytest.param('gzip', 'gzip', gzip.decompress, id='gzip'),
+        pytest.param('brotli', 'br', brotli.decompress, id='brotli'),
+    ],
+)
+async def test_configured_compression_reaches_the_wire_async(
+    httpserver: HTTPServer,
+    compression: HttpCompressionAlgorithm,
+    content_encoding: str,
+    decompress: Callable[[bytes], bytes],
+) -> None:
+    """Async variant of `test_configured_compression_reaches_the_wire_sync`."""
+    captured: list[Request] = []
+    httpserver.expect_request(_ITEMS_PATH, method='POST').respond_with_handler(_capture_body(captured))
+
+    api_url = httpserver.url_for('/').removesuffix('/')
+    client = ApifyClientAsync(token='test_token', api_url=api_url, compression=compression)
+
+    await client.dataset('test_dataset_id').push_items(_LARGE_BODY)
+
+    assert len(captured) == 1
+    assert captured[0].headers['Content-Encoding'] == content_encoding
+    assert json.loads(decompress(captured[0].get_data())) == _LARGE_BODY
diff --git a/tests/unit/test_key_value_store.py b/tests/unit/test_key_value_store.py
index eb2176ef..5142f00c 100644
--- a/tests/unit/test_key_value_store.py
+++ b/tests/unit/test_key_value_store.py
@@ -9,6 +9,7 @@
 from werkzeug import Request, Response
 
 from apify_client import ApifyClient, ApifyClientAsync
+from apify_client._consts import MIN_COMPRESSION_SIZE
 
 if TYPE_CHECKING:
     from collections.abc import Callable
@@ -20,20 +21,25 @@
 _MOCKED_KVS_ID = 'test_kvs_id'
 _RECORD_PATH = f'/v2/key-value-stores/{_MOCKED_KVS_ID}/records/f'
 
+# The client compresses only bodies of at least `MIN_COMPRESSION_SIZE` bytes, so the value is padded past that
+# to keep the compression axis meaningful. A shorter one would go out uncompressed under either algorithm.
+_TEXT_VALUE = 'buffer data' + '.' * MIN_COMPRESSION_SIZE
+_BYTES_VALUE = _TEXT_VALUE.encode('utf-8')
+
 
 class DuckTypedReader:
     """A file-like object that is not an `io.IOBase`, so only duck-typed detection picks it up."""
 
     def read(self) -> bytes:
-        return b'buffer data'
+        return _BYTES_VALUE
 
 
 # The values are built by a factory because reading consumes them, and each case runs once per compression
 # algorithm. Each case is (value factory, expected uploaded body, expected content type).
 _FILE_LIKE_VALUE_CASES = [
-    pytest.param(lambda: io.BytesIO(b'buffer data'), b'buffer data', 'application/octet-stream', id='bytes io'),
-    pytest.param(lambda: io.StringIO('buffer data'), b'buffer data', 'text/plain; charset=utf-8', id='string io'),
-    pytest.param(DuckTypedReader, b'buffer data', 'application/octet-stream', id='duck-typed reader'),
+    pytest.param(lambda: io.BytesIO(_BYTES_VALUE), _BYTES_VALUE, 'application/octet-stream', id='bytes io'),
+    pytest.param(lambda: io.StringIO(_TEXT_VALUE), _BYTES_VALUE, 'text/plain; charset=utf-8', id='string io'),
+    pytest.param(DuckTypedReader, _BYTES_VALUE, 'application/octet-stream', id='duck-typed reader'),
 ]
 
 
diff --git a/tests/unit/test_run_charge.py b/tests/unit/test_run_charge.py
index aa176734..961fc4ca 100644
--- a/tests/unit/test_run_charge.py
+++ b/tests/unit/test_run_charge.py
@@ -1,10 +1,8 @@
 from __future__ import annotations
 
-import gzip
 import json
 from typing import TYPE_CHECKING
 
-import brotli
 import pytest
 from werkzeug import Request, Response
 
@@ -13,29 +11,10 @@
 if TYPE_CHECKING:
     from pytest_httpserver import HTTPServer
 
-    from apify_client.types import HttpCompressionAlgorithm
-
 _MOCKED_RUN_ID = 'test_run_id'
 _CHARGE_PATH = f'/v2/actor-runs/{_MOCKED_RUN_ID}/charge'
 
 
-def _decode_body(request: Request) -> dict:
-    raw = request.get_data()
-    encoding = request.headers.get('Content-Encoding')
-    if encoding == 'br':
-        raw = brotli.decompress(raw)
-    elif encoding == 'gzip':
-        raw = gzip.decompress(raw)
-    return json.loads(raw)
-
-
-@pytest.mark.parametrize(
-    'compression',
-    [
-        pytest.param('gzip', id='gzip'),
-        pytest.param('brotli', id='brotli'),
-    ],
-)
 @pytest.mark.parametrize(
     'count',
     [
@@ -44,12 +23,8 @@ def _decode_body(request: Request) -> dict:
         pytest.param(5, id='five'),
     ],
 )
-def test_run_charge_preserves_count_sync(
-    httpserver: HTTPServer,
-    count: int,
-    compression: HttpCompressionAlgorithm,
-) -> None:
-    """Ensure `count` is sent as-is (in particular, `0` is preserved), regardless of the request-body compression."""
+def test_run_charge_preserves_count_sync(httpserver: HTTPServer, count: int) -> None:
+    """Ensure `count` is sent as-is, in particular that `0` is preserved rather than dropped as falsy."""
     captured_requests: list[Request] = []
 
     def capture_request(request: Request) -> Response:
@@ -59,22 +34,15 @@ def capture_request(request: Request) -> Response:
     httpserver.expect_request(_CHARGE_PATH, method='POST').respond_with_handler(capture_request)
 
     api_url = httpserver.url_for('/').removesuffix('/')
-    client = ApifyClient(token='test_token', api_url=api_url, compression=compression)
+    client = ApifyClient(token='test_token', api_url=api_url)
 
     client.run(_MOCKED_RUN_ID).charge('test-event', count=count)
 
     assert len(captured_requests) == 1
-    body = _decode_body(captured_requests[0])
+    body = json.loads(captured_requests[0].get_data())
     assert body['count'] == count
 
 
-@pytest.mark.parametrize(
-    'compression',
-    [
-        pytest.param('gzip', id='gzip'),
-        pytest.param('brotli', id='brotli'),
-    ],
-)
 @pytest.mark.parametrize(
     'count',
     [
@@ -83,11 +51,7 @@ def capture_request(request: Request) -> Response:
         pytest.param(5, id='five'),
     ],
 )
-async def test_run_charge_preserves_count_async(
-    httpserver: HTTPServer,
-    count: int,
-    compression: HttpCompressionAlgorithm,
-) -> None:
+async def test_run_charge_preserves_count_async(httpserver: HTTPServer, count: int) -> None:
     """Async variant of `test_run_charge_preserves_count_sync`."""
     captured_requests: list[Request] = []
 
@@ -98,10 +62,10 @@ def capture_request(request: Request) -> Response:
     httpserver.expect_request(_CHARGE_PATH, method='POST').respond_with_handler(capture_request)
 
     api_url = httpserver.url_for('/').removesuffix('/')
-    client = ApifyClientAsync(token='test_token', api_url=api_url, compression=compression)
+    client = ApifyClientAsync(token='test_token', api_url=api_url)
 
     await client.run(_MOCKED_RUN_ID).charge('test-event', count=count)
 
     assert len(captured_requests) == 1
-    body = _decode_body(captured_requests[0])
+    body = json.loads(captured_requests[0].get_data())
     assert body['count'] == count