diff --git a/examples/async_example.py b/examples/async_example.py
new file mode 100644
index 0000000..73e67bc
--- /dev/null
+++ b/examples/async_example.py
@@ -0,0 +1,178 @@
+"""Example demonstrating async/coroutine support in Cachier.
+
+This example shows how to use the @cachier decorator with async functions to cache the results of HTTP requests or other
+async operations.
+
+"""
+
+import asyncio
+import time
+from datetime import timedelta
+
+from cachier import cachier
+
+
+# Example 1: Basic async function caching
+@cachier(backend="pickle", stale_after=timedelta(hours=1))
+async def fetch_user_data(user_id: int) -> dict:
+    """Simulate fetching user data from an API."""
+    print(f"  Fetching user {user_id} from API...")
+    await asyncio.sleep(1)  # Simulate network delay
+    return {"id": user_id, "name": f"User{user_id}", "email": f"user{user_id}@example.com"}
+
+
+# Example 2: Async function with memory backend (faster, but not persistent)
+@cachier(backend="memory")
+async def calculate_complex_result(x: int, y: int) -> int:
+    """Simulate a complex calculation."""
+    print(f"  Computing {x} ** {y}...")
+    await asyncio.sleep(0.5)  # Simulate computation time
+    return x**y
+
+
+# Example 3: Async function with stale_after (without next_time for simplicity)
+@cachier(backend="memory", stale_after=timedelta(seconds=3), next_time=False)
+async def get_weather_data(city: str) -> dict:
+    """Simulate fetching weather data with automatic refresh when stale."""
+    print(f"  Fetching weather for {city}...")
+    await asyncio.sleep(0.5)
+    return {"city": city, "temp": 72, "condition": "sunny", "timestamp": time.time()}
+
+
+# Example 4: Real-world HTTP request caching (requires httpx)
+async def demo_http_caching():
+    """Demonstrate caching actual HTTP requests."""
+    print("\n=== HTTP Request Caching Example ===")
+    try:
+        import httpx
+
+        @cachier(backend="pickle", stale_after=timedelta(minutes=5))
+        async def fetch_github_user(username: str) -> dict:
+            """Fetch GitHub user data with caching."""
+            print(f"  Making API request for {username}...")
+            async with httpx.AsyncClient() as client:
+                response = await client.get(f"https://api.github.com/users/{username}")
+                return response.json()
+
+        # First call - makes actual HTTP request
+        start = time.time()
+        user1 = await fetch_github_user("torvalds")
+        duration1 = time.time() - start
+        print(f"  First call took {duration1:.2f}s")
+        user_name = user1.get("name", "N/A")
+        user_repos = user1.get("public_repos", "N/A")
+        print(f"  User: {user_name}, Repos: {user_repos}")
+
+        # Second call - uses cache (much faster)
+        start = time.time()
+        await fetch_github_user("torvalds")
+        duration2 = time.time() - start
+        print(f"  Second call took {duration2:.2f}s (from cache)")
+        if duration2 > 0:
+            print(f"  Cache speedup: {duration1 / duration2:.1f}x")
+        else:
+            print("  Cache speedup: instantaneous (duration too small to measure)")
+
+    except ImportError:
+        print("  (Skipping - httpx not installed. Install with: pip install httpx)")
+
+
+async def main():
+    """Run all async caching examples."""
+    print("=" * 60)
+    print("Cachier Async/Coroutine Support Examples")
+    print("=" * 60)
+
+    # Example 1: Basic async caching
+    print("\n=== Example 1: Basic Async Caching ===")
+    start = time.time()
+    user = await fetch_user_data(42)
+    duration1 = time.time() - start
+    print(f"First call: {user} (took {duration1:.2f}s)")
+
+    start = time.time()
+    user = await fetch_user_data(42)
+    duration2 = time.time() - start
+    print(f"Second call: {user} (took {duration2:.2f}s)")
+    if duration2 > 0:
+        print(f"Speedup: {duration1 / duration2:.1f}x faster!")
+    else:
+        print("Speedup: instantaneous (duration too small to measure)")
+
+    # Example 2: Memory backend
+    print("\n=== Example 2: Memory Backend (Fast, Non-Persistent) ===")
+    start = time.time()
+    result = await calculate_complex_result(2, 20)
+    duration1 = time.time() - start
+    print(f"First call: 2^20 = {result} (took {duration1:.2f}s)")
+
+    start = time.time()
+    result = await calculate_complex_result(2, 20)
+    duration2 = time.time() - start
+    print(f"Second call: 2^20 = {result} (took {duration2:.2f}s)")
+
+    # Example 3: Stale-after
+    print("\n=== Example 3: Stale-After ===")
+    weather = await get_weather_data("San Francisco")
+    print(f"First call: {weather}")
+
+    weather = await get_weather_data("San Francisco")
+    print(f"Second call (cached): {weather}")
+
+    print("Waiting 4 seconds for cache to become stale...")
+    await asyncio.sleep(4)
+
+    weather = await get_weather_data("San Francisco")
+    print(f"Third call (recalculates because stale): {weather}")
+
+    # Example 4: Concurrent requests
+    print("\n=== Example 4: Concurrent Async Requests ===")
+    print("Making 5 concurrent requests...")
+    print("(First 3 are unique and will execute, last 2 are duplicates)")
+    start = time.time()
+    await asyncio.gather(
+        fetch_user_data(1),
+        fetch_user_data(2),
+        fetch_user_data(3),
+        fetch_user_data(1),  # Duplicate - will execute in parallel with first
+        fetch_user_data(2),  # Duplicate - will execute in parallel with second
+    )
+    duration = time.time() - start
+    print(f"All requests completed in {duration:.2f}s")
+
+    # Now test that subsequent calls use cache
+    print("\nMaking the same requests again (should use cache):")
+    start = time.time()
+    await asyncio.gather(
+        fetch_user_data(1),
+        fetch_user_data(2),
+        fetch_user_data(3),
+    )
+    duration2 = time.time() - start
+    print(f"Completed in {duration2:.2f}s - much faster!")
+
+    # Example 5: HTTP caching (if httpx is available)
+    await demo_http_caching()
+
+    # Clean up
+    print("\n=== Cleanup ===")
+    fetch_user_data.clear_cache()
+    calculate_complex_result.clear_cache()
+    get_weather_data.clear_cache()
+    print("All caches cleared!")
+
+    print("\n" + "=" * 60)
+    print("Key Features Demonstrated:")
+    print("  - Async function caching with @cachier decorator")
+    print("  - Multiple backends (pickle, memory)")
+    print("  - Automatic cache invalidation (stale_after)")
+    print("  - Concurrent request handling")
+    print("  - Significant performance improvements")
+    print("\nNote: For async functions, concurrent calls with the same")
+    print("arguments will execute in parallel initially. Subsequent calls")
+    print("will use the cached result for significant speedup.")
+    print("=" * 60)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())
diff --git a/pyproject.toml b/pyproject.toml
index 7d81cfb..aa0a9a2 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -147,8 +147,8 @@ lint.mccabe.max-complexity = 10
 [tool.docformatter]
 recursive = true
 # some docstring start with r"""
-wrap-summaries = 79
-wrap-descriptions = 79
+wrap-summaries = 120
+wrap-descriptions = 120
 blank = true
 
 # === Testing ===
@@ -178,6 +178,7 @@ markers = [
   "redis: test the Redis core",
   "sql: test the SQL core",
   "maxage: test the max_age functionality",
+  "asyncio: marks tests as async",
 ]
 
 # --- coverage ---
diff --git a/src/cachier/core.py b/src/cachier/core.py
index fa21176..85870b9 100644
--- a/src/cachier/core.py
+++ b/src/cachier/core.py
@@ -7,6 +7,7 @@
 # http://www.opensource.org/licenses/MIT-license
 # Copyright (c) 2016, Shay Palachy <shaypal5@gmail.com>
 
+import asyncio
 import inspect
 import os
 import threading
@@ -56,6 +57,14 @@ def _function_thread(core, key, func, args, kwds):
         print(f"Function call failed with the following exception:\n{exc}")
 
 
+async def _function_thread_async(core, key, func, args, kwds):
+    try:
+        func_res = await func(*args, **kwds)
+        core.set_entry(key, func_res)
+    except BaseException as exc:
+        print(f"Function call failed with the following exception:\n{exc}")
+
+
 def _calc_entry(core, key, func, args, kwds, printer=lambda *_: None) -> Optional[Any]:
     core.mark_entry_being_calculated(key)
     try:
@@ -68,6 +77,18 @@ def _calc_entry(core, key, func, args, kwds, printer=lambda *_: None) -> Optiona
         core.mark_entry_not_calculated(key)
 
 
+async def _calc_entry_async(core, key, func, args, kwds, printer=lambda *_: None) -> Optional[Any]:
+    core.mark_entry_being_calculated(key)
+    try:
+        func_res = await func(*args, **kwds)
+        stored = core.set_entry(key, func_res)
+        if not stored:
+            printer("Result exceeds entry_size_limit; not cached")
+        return func_res
+    finally:
+        core.mark_entry_not_calculated(key)
+
+
 def _convert_args_kwargs(func, _is_method: bool, args: tuple, kwds: dict) -> dict:
     """Convert mix of positional and keyword arguments to aggregated kwargs."""
     # unwrap if the function is functools.partial
@@ -390,13 +411,108 @@ def _call(*args, max_age: Optional[timedelta] = None, **kwds):
             _print("No entry found. No current calc. Calling like a boss.")
             return _calc_entry(core, key, func, args, kwds, _print)
 
+        async def _call_async(*args, max_age: Optional[timedelta] = None, **kwds):
+            # NOTE: For async functions, wait_for_calc_timeout is not honored.
+            # Instead of blocking the event loop waiting for concurrent
+            # calculations, async functions will recalculate in parallel.
+            # This avoids deadlocks and maintains async efficiency.
+            nonlocal allow_none, last_cleanup
+            _allow_none = _update_with_defaults(allow_none, "allow_none", kwds)
+            # print('Inside async wrapper for {}.'.format(func.__name__))
+            ignore_cache = _pop_kwds_with_deprecation(kwds, "ignore_cache", False)
+            overwrite_cache = _pop_kwds_with_deprecation(kwds, "overwrite_cache", False)
+            verbose = _pop_kwds_with_deprecation(kwds, "verbose_cache", False)
+            ignore_cache = kwds.pop("cachier__skip_cache", ignore_cache)
+            overwrite_cache = kwds.pop("cachier__overwrite_cache", overwrite_cache)
+            verbose = kwds.pop("cachier__verbose", verbose)
+            _stale_after = _update_with_defaults(stale_after, "stale_after", kwds)
+            _next_time = _update_with_defaults(next_time, "next_time", kwds)
+            _cleanup_flag = _update_with_defaults(cleanup_stale, "cleanup_stale", kwds)
+            _cleanup_interval_val = _update_with_defaults(cleanup_interval, "cleanup_interval", kwds)
+            # merge args expanded as kwargs and the original kwds
+            kwargs = _convert_args_kwargs(func, _is_method=core.func_is_method, args=args, kwds=kwds)
+
+            if _cleanup_flag:
+                now = datetime.now()
+                with cleanup_lock:
+                    if now - last_cleanup >= _cleanup_interval_val:
+                        last_cleanup = now
+                        _get_executor().submit(core.delete_stale_entries, _stale_after)
+
+            _print = print if verbose else lambda x: None
+
+            # Check current global caching state dynamically
+            from .config import _global_params
+
+            if ignore_cache or not _global_params.caching_enabled:
+                return await func(args[0], **kwargs) if core.func_is_method else await func(**kwargs)
+            key, entry = core.get_entry((), kwargs)
+            if overwrite_cache:
+                result = await _calc_entry_async(core, key, func, args, kwds, _print)
+                return result
+            if entry is None or (not entry._completed and not entry._processing):
+                _print("No entry found. No current calc. Calling like a boss.")
+                result = await _calc_entry_async(core, key, func, args, kwds, _print)
+                return result
+            _print("Entry found.")
+            if _allow_none or entry.value is not None:
+                _print("Cached result found.")
+                now = datetime.now()
+                max_allowed_age = _stale_after
+                nonneg_max_age = True
+                if max_age is not None:
+                    if max_age < ZERO_TIMEDELTA:
+                        _print("max_age is negative. Cached result considered stale.")
+                        nonneg_max_age = False
+                    else:
+                        assert max_age is not None  # noqa: S101
+                        max_allowed_age = min(_stale_after, max_age)
+                # note: if max_age < 0, we always consider a value stale
+                if nonneg_max_age and (now - entry.time <= max_allowed_age):
+                    _print("And it is fresh!")
+                    return entry.value
+                _print("But it is stale... :(")
+                if _next_time:
+                    _print("Async calc and return stale")
+                    # Mark entry as being calculated then immediately unmark
+                    # This matches sync behavior and ensures entry exists
+                    # Background task will update cache when complete
+                    core.mark_entry_being_calculated(key)
+                    # Use asyncio.create_task for background execution
+                    asyncio.create_task(_function_thread_async(core, key, func, args, kwds))
+                    core.mark_entry_not_calculated(key)
+                    return entry.value
+                _print("Calling decorated function and waiting")
+                result = await _calc_entry_async(core, key, func, args, kwds, _print)
+                return result
+            if entry._processing:
+                msg = "No value but being calculated. Recalculating"
+                _print(f"{msg} (async - no wait).")
+                # For async, don't wait - just recalculate
+                # This avoids blocking the event loop
+                result = await _calc_entry_async(core, key, func, args, kwds, _print)
+                return result
+            _print("No entry found. No current calc. Calling like a boss.")
+            return await _calc_entry_async(core, key, func, args, kwds, _print)
+
         # MAINTAINER NOTE: The main function wrapper is now a standard function
         # that passes *args and **kwargs to _call. This ensures that user
         # arguments are not shifted, and max_age is only settable via keyword
         # argument.
-        @wraps(func)
-        def func_wrapper(*args, **kwargs):
-            return _call(*args, **kwargs)
+        # For async functions, we create an async wrapper that calls
+        # _call_async.
+        is_coroutine = inspect.iscoroutinefunction(func)
+
+        if is_coroutine:
+
+            @wraps(func)
+            async def func_wrapper(*args, **kwargs):
+                return await _call_async(*args, **kwargs)
+        else:
+
+            @wraps(func)
+            def func_wrapper(*args, **kwargs):
+                return _call(*args, **kwargs)
 
         def _clear_cache():
             """Clear the cache."""
diff --git a/src/cachier/cores/base.py b/src/cachier/cores/base.py
index a22ddda..ce1bda7 100644
--- a/src/cachier/cores/base.py
+++ b/src/cachier/cores/base.py
@@ -29,9 +29,8 @@ class RecalculationNeeded(Exception):
 def _get_func_str(func: Callable) -> str:
     """Return a string identifier for the function (module + name).
 
-    We accept Any here because static analysis can't always prove that the
-    runtime object will have __module__ and __name__, but at runtime the
-    decorated functions always do.
+    We accept Any here because static analysis can't always prove that the runtime object will have __module__ and
+    __name__, but at runtime the decorated functions always do.
 
     """
     return f".{func.__module__}.{func.__name__}"
@@ -52,8 +51,7 @@ def __init__(
     def set_func(self, func):
         """Set the function this core will use.
 
-        This has to be set before any method is called. Also determine if the
-        function is an object method.
+        This has to be set before any method is called. Also determine if the function is an object method.
 
         """
         # unwrap if the function is functools.partial
@@ -70,8 +68,7 @@ def get_key(self, args, kwds):
     def get_entry(self, args, kwds) -> Tuple[str, Optional[CacheEntry]]:
         """Get entry based on given arguments.
 
-        Return the result mapped to the given arguments in this core's cache,
-        if such a mapping exists.
+        Return the result mapped to the given arguments in this core's cache, if such a mapping exists.
 
         """
         key = self.get_key(args, kwds)
diff --git a/tests/requirements.txt b/tests/requirements.txt
index d34de0b..7829727 100644
--- a/tests/requirements.txt
+++ b/tests/requirements.txt
@@ -3,6 +3,7 @@
 pytest
 coverage
 pytest-cov
+pytest-asyncio
 birch
 # to be able to run `python setup.py checkdocs`
 collective.checkdocs
diff --git a/tests/test_async_core.py b/tests/test_async_core.py
new file mode 100644
index 0000000..e62775d
--- /dev/null
+++ b/tests/test_async_core.py
@@ -0,0 +1,979 @@
+"""Tests for async/coroutine support in Cachier."""
+
+import asyncio
+from datetime import timedelta
+from time import sleep, time
+
+import pytest
+
+from cachier import cachier
+
+# =============================================================================
+# Basic Async Caching Tests
+# =============================================================================
+
+
+class TestBasicAsyncCaching:
+    """Tests for basic async caching functionality."""
+
+    @pytest.mark.memory
+    @pytest.mark.asyncio
+    async def test_memory(self):
+        """Test basic async caching with memory backend."""
+
+        @cachier(backend="memory")
+        async def async_func(x):
+            await asyncio.sleep(0.1)
+            return x * 2
+
+        async_func.clear_cache()
+
+        # First call should execute
+        result1 = await async_func(5)
+        assert result1 == 10
+
+        # Second call should use cache
+        start = time()
+        result2 = await async_func(5)
+        end = time()
+        assert result2 == 10
+        assert end - start < 0.05  # Should be much faster than 0.1s
+
+        async_func.clear_cache()
+
+    @pytest.mark.pickle
+    @pytest.mark.asyncio
+    async def test_pickle(self):
+        """Test basic async caching with pickle backend."""
+
+        @cachier(backend="pickle")
+        async def async_func(x):
+            await asyncio.sleep(0.1)
+            return x * 3
+
+        async_func.clear_cache()
+
+        # First call should execute
+        result1 = await async_func(7)
+        assert result1 == 21
+
+        # Second call should use cache
+        start = time()
+        result2 = await async_func(7)
+        end = time()
+        assert result2 == 21
+        assert end - start < 0.05  # Should be much faster than 0.1s
+
+        async_func.clear_cache()
+
+
+# =============================================================================
+# Stale Cache Tests
+# =============================================================================
+
+
+@pytest.mark.memory
+@pytest.mark.asyncio
+class TestStaleCache:
+    """Tests for stale_after and next_time functionality."""
+
+    async def test_recalculates_after_expiry(self):
+        """Test that stale_after causes recalculation after expiry."""
+        call_count = 0
+
+        @cachier(
+            backend="memory",
+            stale_after=timedelta(seconds=0.5),
+            next_time=False,
+        )
+        async def async_func(x):
+            nonlocal call_count
+            call_count += 1
+            await asyncio.sleep(0.1)
+            return x * 2
+
+        async_func.clear_cache()
+        call_count = 0
+
+        # First call
+        result1 = await async_func(5)
+        assert result1 == 10
+        assert call_count == 1
+
+        # Wait for cache to become stale
+        await asyncio.sleep(0.6)
+
+        # Second call - should recalculate
+        result2 = await async_func(5)
+        assert result2 == 10
+        assert call_count == 2
+
+        async_func.clear_cache()
+
+    async def test_uses_cache_before_expiry(self):
+        """Test that cache is used before stale_after expiry."""
+        call_count = 0
+
+        @cachier(backend="memory", stale_after=timedelta(seconds=1), next_time=False)
+        async def async_func(x):
+            nonlocal call_count
+            call_count += 1
+            await asyncio.sleep(0.1)
+            return x * 2
+
+        async_func.clear_cache()
+        call_count = 0
+
+        # First call
+        result1 = await async_func(5)
+        assert result1 == 10
+        assert call_count == 1
+
+        # Second call - should use cache (no additional call)
+        previous_call_count = call_count
+        result2 = await async_func(5)
+        assert result2 == 10
+        assert call_count == previous_call_count  # Verify cache was used
+
+        async_func.clear_cache()
+
+    async def test_next_time_returns_stale_and_updates_background(self):
+        """Test next_time=True returns stale value and updates in bg."""
+        call_count = 0
+
+        @cachier(
+            backend="memory",
+            stale_after=timedelta(seconds=0.5),
+            next_time=True,
+        )
+        async def async_func(x):
+            nonlocal call_count
+            call_count += 1
+            await asyncio.sleep(0.1)
+            return call_count * 10
+
+        async_func.clear_cache()
+        call_count = 0
+
+        # First call
+        result1 = await async_func(5)
+        assert result1 == 10
+        assert call_count == 1
+
+        # Wait for cache to become stale
+        await asyncio.sleep(0.6)
+
+        # Second call - should return stale value and trigger background update
+        result2 = await async_func(5)
+        assert result2 == 10  # Still returns old value
+
+        # Wait for background calculation to complete
+        await asyncio.sleep(0.5)
+
+        # Third call - should return new value
+        result3 = await async_func(5)
+        assert result3 == 20  # New value from background calculation
+
+        async_func.clear_cache()
+
+
+# =============================================================================
+# Cache Control Tests
+# =============================================================================
+
+
+@pytest.mark.memory
+@pytest.mark.asyncio
+class TestCacheControl:
+    """Tests for cache control parameters - skip_cache & overwrite_cache."""
+
+    async def test_skip_cache(self):
+        """Test async caching with cachier__skip_cache parameter."""
+        call_count = 0
+
+        @cachier(backend="memory")
+        async def async_func(x):
+            nonlocal call_count
+            call_count += 1
+            await asyncio.sleep(0.1)
+            return call_count * 10
+
+        async_func.clear_cache()
+        call_count = 0
+
+        # First call
+        result1 = await async_func(5)
+        assert result1 == 10
+
+        # Second call with skip_cache
+        result2 = await async_func(5, cachier__skip_cache=True)
+        assert result2 == 20
+
+        # Third call - should use cache from first call
+        result3 = await async_func(5)
+        assert result3 == 10
+
+        async_func.clear_cache()
+
+    async def test_overwrite_cache(self):
+        """Test async caching with cachier__overwrite_cache parameter."""
+        call_count = 0
+
+        @cachier(backend="memory")
+        async def async_func(x):
+            nonlocal call_count
+            call_count += 1
+            await asyncio.sleep(0.1)
+            return call_count * 10
+
+        async_func.clear_cache()
+        call_count = 0
+
+        # First call
+        result1 = await async_func(5)
+        assert result1 == 10
+
+        # Second call with overwrite_cache
+        result2 = await async_func(5, cachier__overwrite_cache=True)
+        assert result2 == 20
+
+        # Third call - should use new cached value
+        result3 = await async_func(5)
+        assert result3 == 20
+
+        async_func.clear_cache()
+
+
+# =============================================================================
+# Class Method Tests
+# =============================================================================
+
+
+@pytest.mark.memory
+@pytest.mark.asyncio
+class TestAsyncMethod:
+    """Tests for async caching on class methods."""
+
+    async def test_caches_result(self):
+        """Test async caching on class methods returns cached result."""
+
+        class MyClass:
+            def __init__(self, value):
+                self.value = value
+
+            @cachier(backend="memory")
+            async def async_method(self, x):
+                await asyncio.sleep(0.1)
+                return x * self.value
+
+        obj1 = MyClass(2)
+
+        obj1.async_method.clear_cache()
+
+        # First call on obj1
+        result1 = await obj1.async_method(5)
+        assert result1 == 10
+
+        # Second call on obj1 - should use cache
+        start = time()
+        result2 = await obj1.async_method(5)
+        end = time()
+        assert result2 == 10
+        assert end - start < 0.05
+
+        obj1.async_method.clear_cache()
+
+    async def test_shares_cache_across_instances(self):
+        """Test that async method cache is shared across instances."""
+
+        class MyClass:
+            def __init__(self, value):
+                self.value = value
+
+            @cachier(backend="memory")
+            async def async_method(self, x):
+                await asyncio.sleep(0.1)
+                return x * self.value
+
+        obj1 = MyClass(2)
+        obj2 = MyClass(3)
+
+        obj1.async_method.clear_cache()
+
+        # First call on obj1
+        result1 = await obj1.async_method(5)
+        assert result1 == 10
+
+        # Call on obj2 with same argument - should also use cache
+        # (because cache is based on method arguments, not instance)
+        result2 = await obj2.async_method(5)
+        assert result2 == 10  # Returns cached value from obj1
+
+        obj1.async_method.clear_cache()
+
+
+# =============================================================================
+# Sync Function Compatibility Tests
+# =============================================================================
+
+
+class TestSyncCompatibility:
+    """Tests to ensure sync functions still work."""
+
+    @pytest.mark.memory
+    def test_still_works(self):
+        """Ensure sync functions still work after adding async support."""
+
+        @cachier(backend="memory")
+        def sync_func(x):
+            sleep(0.1)
+            return x * 2
+
+        sync_func.clear_cache()
+
+        # First call
+        result1 = sync_func(5)
+        assert result1 == 10
+
+        # Second call should use cache
+        start = time()
+        result2 = sync_func(5)
+        end = time()
+        assert result2 == 10
+        assert end - start < 0.05
+
+        sync_func.clear_cache()
+
+
+# =============================================================================
+# Argument Handling Tests
+# =============================================================================
+
+
+class TestArgumentHandling:
+    """Tests for different argument types and patterns."""
+
+    @pytest.mark.parametrize(
+        ("args", "kwargs", "expected"),
+        [
+            ((1, 2), {}, 13),  # positional args
+            ((1,), {"y": 2}, 13),  # keyword args
+            ((1, 2), {"z": 5}, 8),  # different default override
+        ],
+    )
+    @pytest.mark.memory
+    @pytest.mark.asyncio
+    async def test_different_types(self, args, kwargs, expected):
+        """Test async caching with different argument types."""
+
+        @cachier(backend="memory")
+        async def async_func(x, y, z=10):
+            await asyncio.sleep(0.1)
+            return x + y + z
+
+        async_func.clear_cache()
+
+        result = await async_func(*args, **kwargs)
+        assert result == expected
+
+        async_func.clear_cache()
+
+
+# =============================================================================
+# Max Age Tests
+# =============================================================================
+
+
+@pytest.mark.memory
+@pytest.mark.asyncio
+@pytest.mark.maxage
+class TestMaxAge:
+    """Tests for max_age parameter functionality."""
+
+    async def test_recalculates_when_expired(self):
+        """Test that max_age causes recalculation when cache is too old."""
+        call_count = 0
+
+        @cachier(backend="memory", stale_after=timedelta(days=1))
+        async def async_func(x):
+            nonlocal call_count
+            call_count += 1
+            await asyncio.sleep(0.1)
+            return x * 2
+
+        async_func.clear_cache()
+        call_count = 0
+
+        # First call
+        result1 = await async_func(5)
+        assert result1 == 10
+        assert call_count == 1
+
+        # Wait a bit
+        await asyncio.sleep(0.5)
+
+        # Second call with max_age - should recalculate because cache is older
+        # than max_age
+        result2 = await async_func(5, max_age=timedelta(milliseconds=100))
+        assert result2 == 10
+        assert call_count == 2
+
+        async_func.clear_cache()
+
+    async def test_uses_cache_when_fresh(self):
+        """Test that cache is used when within max_age."""
+        call_count = 0
+
+        @cachier(backend="memory", stale_after=timedelta(days=1))
+        async def async_func(x):
+            nonlocal call_count
+            call_count += 1
+            await asyncio.sleep(0.1)
+            return x * 2
+
+        async_func.clear_cache()
+        call_count = 0
+
+        # First call
+        result1 = await async_func(5)
+        assert result1 == 10
+        assert call_count == 1
+
+        # Second call with max_age - should use cache
+        previous_call_count = call_count
+        result2 = await async_func(5, max_age=timedelta(seconds=10))
+        assert result2 == 10
+        assert call_count == previous_call_count  # No additional call
+
+        async_func.clear_cache()
+
+    async def test_negative_max_age_forces_recalculation(self):
+        """Test that negative max_age forces recalculation."""
+        call_count = 0
+
+        @cachier(backend="memory", stale_after=timedelta(days=1))
+        async def async_func(x):
+            nonlocal call_count
+            call_count += 1
+            await asyncio.sleep(0.1)
+            return x * 2
+
+        async_func.clear_cache()
+        call_count = 0
+
+        # First call
+        result1 = await async_func(5)
+        assert result1 == 10
+        assert call_count == 1
+
+        # Second call with negative max_age - should recalculate
+        result2 = await async_func(5, max_age=timedelta(seconds=-1))
+        assert result2 == 10
+        assert call_count == 2
+
+        async_func.clear_cache()
+
+
+# =============================================================================
+# Concurrent Access Tests
+# =============================================================================
+
+
+@pytest.mark.memory
+@pytest.mark.asyncio
+class TestConcurrentAccess:
+    """Tests for concurrent async call behavior."""
+
+    async def test_calls_execute_in_parallel(self):
+        """Test that concurrent async calls execute in parallel."""
+        call_count = 0
+
+        @cachier(backend="memory")
+        async def async_func(x):
+            nonlocal call_count
+            call_count += 1
+            await asyncio.sleep(0.2)
+            return x * 2
+
+        async_func.clear_cache()
+        call_count = 0
+
+        # First concurrent calls - all will execute in parallel
+        results1 = await asyncio.gather(
+            async_func(5),
+            async_func(5),
+            async_func(5),
+        )
+        assert all(r == 10 for r in results1)
+        # All three calls executed
+        assert call_count == 3
+
+        async_func.clear_cache()
+
+    async def test_consequent_calls_use_cache(self):
+        """Test that calls after caching use cached value."""
+        call_count = 0
+
+        @cachier(backend="memory")
+        async def async_func(x):
+            nonlocal call_count
+            call_count += 1
+            await asyncio.sleep(0.2)
+            return x * 2
+
+        async_func.clear_cache()
+        call_count = 0
+
+        # First call to populate cache
+        await async_func(5)
+        assert call_count == 1
+
+        # Subsequent calls should use cache
+        call_count = 0
+        results2 = await asyncio.gather(
+            async_func(5),
+            async_func(5),
+            async_func(5),
+        )
+        assert all(r == 10 for r in results2)
+        assert call_count == 0  # No new calls, all from cache
+
+        async_func.clear_cache()
+
+    async def test_stale_entry_being_processed_with_next_time(self):
+        """Test concurrent calls with stale cache and next_time=True return stale values.
+
+        When cache is stale and next_time=True, concurrent calls should return the stale value while background
+        recalculation happens.
+
+        """
+        call_count = 0
+
+        @cachier(backend="memory", stale_after=timedelta(seconds=1), next_time=True)
+        async def slow_async_func(x):
+            nonlocal call_count
+            call_count += 1
+            await asyncio.sleep(1.0)  # Long enough to create processing overlap
+            return call_count * 10
+
+        slow_async_func.clear_cache()
+        call_count = 0
+
+        # First call - populate cache
+        result1 = await slow_async_func(5)
+        assert result1 == 10
+        assert call_count == 1
+
+        # Wait for cache to become stale
+        await asyncio.sleep(1.5)
+
+        # Start a slow recalculation in background (don't await it yet)
+        task1 = asyncio.create_task(slow_async_func(5))
+
+        # Give it a moment to mark entry as being processed
+        await asyncio.sleep(0.1)
+
+        # Now make another call while first one is still processing
+        # This should return the stale value because entry._processing=True and next_time=True
+        result2 = await slow_async_func(5)
+        assert result2 == 10  # Should return stale value
+
+        # Wait for background task to complete
+        await task1
+
+        # Wait enough time for the background update to complete and cache to be updated
+        await asyncio.sleep(1.5)
+
+        # Next call should get an updated value (could be 20 or 30 depending on background tasks)
+        result3 = await slow_async_func(5)
+        assert result3 > 10  # Should be updated from background
+
+        slow_async_func.clear_cache()
+
+
+# =============================================================================
+# None Value Handling Tests
+# =============================================================================
+
+
+@pytest.mark.memory
+@pytest.mark.asyncio
+class TestNoneHandling:
+    """Tests for allow_none parameter behavior."""
+
+    async def test_not_cached_by_default(self):
+        """Test that None values are not cached when allow_none=False."""
+        call_count = 0
+
+        @cachier(backend="memory")
+        async def async_func(x):
+            nonlocal call_count
+            call_count += 1
+            await asyncio.sleep(0.1)
+            return None if x == 0 else x * 2
+
+        async_func.clear_cache()
+        call_count = 0
+
+        # First call returning None - should not be cached
+        result1 = await async_func(0)
+        assert result1 is None
+        assert call_count == 1
+
+        # Second call with same args - should recalculate (None not cached)
+        result2 = await async_func(0)
+        assert result2 is None
+        assert call_count == 2
+
+        async_func.clear_cache()
+
+    async def test_cached_when_allowed(self):
+        """Test that None values are cached when allow_none=True."""
+        call_count = 0
+
+        @cachier(backend="memory", allow_none=True)
+        async def async_func(x):
+            nonlocal call_count
+            call_count += 1
+            await asyncio.sleep(0.1)
+            return None if x == 0 else x * 2
+
+        async_func.clear_cache()
+        call_count = 0
+
+        # First call returning None - should be cached
+        result1 = await async_func(0)
+        assert result1 is None
+        assert call_count == 1
+
+        # Second call with same args - should use cached None
+        previous_call_count = call_count
+        result2 = await async_func(0)
+        assert result2 is None
+        assert call_count == previous_call_count  # No additional call
+
+        async_func.clear_cache()
+
+    async def test_non_none_cached_with_allow_none_false(self):
+        """Test that non-None values are cached even when allow_none=False."""
+        call_count = 0
+
+        @cachier(backend="memory")
+        async def async_func(x):
+            nonlocal call_count
+            call_count += 1
+            await asyncio.sleep(0.1)
+            return None if x == 0 else x * 2
+
+        async_func.clear_cache()
+        call_count = 0
+
+        # Call with non-None result - should be cached
+        result1 = await async_func(5)
+        assert result1 == 10
+        assert call_count == 1
+
+        # Call again - should use cache
+        previous_call_count = call_count
+        result2 = await async_func(5)
+        assert result2 == 10
+        assert call_count == previous_call_count  # No additional call
+
+        async_func.clear_cache()
+
+
+# =============================================================================
+# Additional Coverage Tests
+# =============================================================================
+
+
+@pytest.mark.memory
+@pytest.mark.asyncio
+class TestAsyncVerboseMode:
+    """Tests for verbose_cache parameter with async functions."""
+
+    async def test_verbose_cache_parameter(self, capsys):
+        """Test verbose_cache parameter prints debug info."""
+        import warnings
+
+        call_count = 0
+
+        @cachier(backend="memory")
+        async def async_func(x):
+            nonlocal call_count
+            call_count += 1
+            await asyncio.sleep(0.1)
+            return x * 2
+
+        async_func.clear_cache()
+        call_count = 0
+
+        # First call with verbose=True (deprecated but still works)
+        with warnings.catch_warnings():
+            warnings.simplefilter("ignore", DeprecationWarning)
+            result1 = await async_func(5, verbose_cache=True)
+        assert result1 == 10
+        captured = capsys.readouterr()
+        assert "No entry found" in captured.out or "Calling" in captured.out
+
+        # Second call with verbose=True
+        with warnings.catch_warnings():
+            warnings.simplefilter("ignore", DeprecationWarning)
+            result2 = await async_func(5, verbose_cache=True)
+        assert result2 == 10
+        captured = capsys.readouterr()
+        assert "Entry found" in captured.out or "Cached result" in captured.out
+
+        async_func.clear_cache()
+
+    async def test_cachier_verbose_kwarg(self, capsys):
+        """Test cachier__verbose keyword argument."""
+
+        @cachier(backend="memory")
+        async def async_func(x):
+            await asyncio.sleep(0.1)
+            return x * 3
+
+        async_func.clear_cache()
+
+        # Use cachier__verbose keyword
+        result = await async_func(7, cachier__verbose=True)
+        assert result == 21
+        captured = capsys.readouterr()
+        assert len(captured.out) > 0  # Should have printed something
+
+        async_func.clear_cache()
+
+
+class TestAsyncGlobalCachingControl:
+    """Tests for global caching enable/disable with async functions."""
+
+    @pytest.mark.memory
+    @pytest.mark.asyncio
+    async def test_disable_caching_globally(self):
+        """Test disabling caching globally affects async functions."""
+        import cachier
+
+        call_count = 0
+
+        @cachier.cachier(backend="memory")
+        async def async_func(x):
+            nonlocal call_count
+            call_count += 1
+            await asyncio.sleep(0.1)
+            return x * 2
+
+        async_func.clear_cache()
+        call_count = 0
+
+        # Enable caching (default)
+        cachier.enable_caching()
+
+        # First call - should cache
+        result1 = await async_func(5)
+        assert result1 == 10
+        assert call_count == 1
+
+        # Second call - should use cache
+        result2 = await async_func(5)
+        assert result2 == 10
+        assert call_count == 1
+
+        # Disable caching
+        cachier.disable_caching()
+
+        # Third call - should not use cache
+        result3 = await async_func(5)
+        assert result3 == 10
+        assert call_count == 2
+
+        # Fourth call - still should not use cache
+        result4 = await async_func(5)
+        assert result4 == 10
+        assert call_count == 3
+
+        # Re-enable caching
+        cachier.enable_caching()
+
+        async_func.clear_cache()
+
+
+class TestAsyncCleanupStale:
+    """Tests for cleanup_stale functionality with async functions."""
+
+    @pytest.mark.memory
+    @pytest.mark.asyncio
+    async def test_cleanup_stale_entries(self):
+        """Test that stale entries are cleaned up with cleanup_stale=True."""
+        call_count = 0
+
+        @cachier(
+            backend="memory",
+            stale_after=timedelta(seconds=1),
+            cleanup_stale=True,
+            cleanup_interval=timedelta(milliseconds=100),
+        )
+        async def async_func(x):
+            nonlocal call_count
+            call_count += 1
+            await asyncio.sleep(0.1)
+            return x * 2
+
+        async_func.clear_cache()
+        call_count = 0
+
+        # First call
+        result1 = await async_func(5)
+        assert result1 == 10
+        assert call_count == 1
+
+        # Wait for stale
+        await asyncio.sleep(1.5)
+
+        # Second call - triggers cleanup in background
+        result2 = await async_func(5)
+        assert result2 == 10
+        assert call_count == 2
+
+        # Give cleanup time to run
+        await asyncio.sleep(0.5)
+
+        async_func.clear_cache()
+
+
+@pytest.mark.memory
+@pytest.mark.asyncio
+class TestAsyncProcessingEntry:
+    """Tests for entry being processed scenarios with async functions."""
+
+    async def test_entry_processing_without_value(self):
+        """Test async recalculation when entry is processing but has no value."""
+        call_count = 0
+
+        @cachier(backend="memory")
+        async def async_func(x):
+            nonlocal call_count
+            call_count += 1
+            await asyncio.sleep(0.3)
+            return x * 2
+
+        async_func.clear_cache()
+        call_count = 0
+
+        # Launch concurrent calls - they should all execute
+        results = await asyncio.gather(
+            async_func(10),
+            async_func(10),
+            async_func(10),
+        )
+
+        assert all(r == 20 for r in results)
+        # All three should have executed since async doesn't wait
+        assert call_count == 3
+
+        async_func.clear_cache()
+
+    async def test_stale_entry_processing_recalculates(self):
+        """Test that stale entry being processed causes recalculation."""
+        call_count = 0
+
+        @cachier(backend="memory", stale_after=timedelta(seconds=1))
+        async def async_func(x):
+            nonlocal call_count
+            call_count += 1
+            await asyncio.sleep(0.5)
+            return call_count * 10
+
+        async_func.clear_cache()
+        call_count = 0
+
+        # First call
+        result1 = await async_func(5)
+        assert result1 == 10
+        assert call_count == 1
+
+        # Wait for stale
+        await asyncio.sleep(1.5)
+
+        # Launch concurrent calls on stale entry
+        # Both should recalculate (no waiting in async)
+        await asyncio.gather(
+            async_func(5),
+            async_func(5),
+        )
+
+        # Both should have executed
+        assert call_count >= 2
+
+        async_func.clear_cache()
+
+
+# =============================================================================
+# Exception Handling and Edge Cases
+# =============================================================================
+
+
+@pytest.mark.memory
+@pytest.mark.asyncio
+class TestAsyncExceptionHandling:
+    """Tests for exception handling in async background tasks."""
+
+    async def test_function_thread_async_exception_handling(self, capsys):
+        """Test that exceptions in background async tasks are caught and printed."""
+        exception_raised = False
+
+        @cachier(backend="memory", stale_after=timedelta(seconds=1), next_time=True)
+        async def async_func_that_fails(x):
+            nonlocal exception_raised
+            await asyncio.sleep(0.1)
+            if exception_raised:
+                raise ValueError("Intentional test error in background")
+            return x * 2
+
+        async_func_that_fails.clear_cache()
+
+        # First call with valid value
+        result1 = await async_func_that_fails(5)
+        assert result1 == 10
+
+        # Wait for stale
+        await asyncio.sleep(1.5)
+
+        # Set flag to raise exception in next call
+        exception_raised = True
+
+        # Call again - should return stale value and update in background
+        # Background task will fail and exception should be caught and printed
+        result2 = await async_func_that_fails(5)
+        assert result2 == 10  # Returns stale value
+
+        # Wait for background task to complete and fail
+        await asyncio.sleep(0.5)
+
+        # Check that exception was caught and printed in _function_thread_async
+        captured = capsys.readouterr()
+        assert "Function call failed with the following exception" in captured.out
+        assert "Intentional test error in background" in captured.out
+
+        async_func_that_fails.clear_cache()
+
+    async def test_entry_size_limit_exceeded_async(self, capsys):
+        """Test that exceeding entry_size_limit prints a message."""
+
+        @cachier(backend="memory", entry_size_limit=10)  # Very small limit
+        async def async_func_large_result(x):
+            await asyncio.sleep(0.1)
+            # Return a large result that exceeds 10 bytes
+            return "x" * 1000
+
+        async_func_large_result.clear_cache()
+
+        # Call function with cachier__verbose=True - result should exceed size limit
+        result = await async_func_large_result(5, cachier__verbose=True)
+        assert len(result) == 1000
+
+        # Check that the size limit message was printed
+        captured = capsys.readouterr()
+        assert "Result exceeds entry_size_limit; not cached" in captured.out
+
+        async_func_large_result.clear_cache()
diff --git a/tests/test_caching_regression.py b/tests/test_caching_regression.py
index 9c144bd..82f857c 100644
--- a/tests/test_caching_regression.py
+++ b/tests/test_caching_regression.py
@@ -1,7 +1,6 @@
 """Test for caching enable/disable regression issue.
 
-This test ensures that decorators defined when caching is disabled can still be
-enabled later via enable_caching().
+This test ensures that decorators defined when caching is disabled can still be enabled later via enable_caching().
 
 """