diff --git a/CHANGELOG.md b/CHANGELOG.md
index 51ffc2d..c501777 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,3 +1,26 @@
+# 0.5.0
+
+## Performance
+
+- perf: values are no longer run through `:erlang.term_to_binary/1` for adapters that store Erlang terms natively (`Cache.ETS`, `Cache.Agent`, `Cache.PersistentTerm`, `Cache.ConCache`, `Cache.Counter`). The encode/decode round trip was pure overhead on those adapters, and the decode dominated the lookup it was attached to. On a 500k-entry ETS table, `get/1` of a ~10KB body drops from **29,981 ns to 6,211 ns (4.8x)** and `put/2` from **23,542 ns to 10,804 ns (2.2x)**. The win holds across payload sizes — a small map goes from 2,105 ns to 302 ns (7.0x).
+- `Cache.PersistentTerm` now hands out the stored term itself on every read, restoring the zero-copy property the adapter exists for.
+
+## Features
+
+- feat: added the optional `c:Cache.native_term_storage?/1` callback, letting an adapter declare that it stores Erlang terms natively. It is resolved at compile time, so there is no runtime branch on the read or write path. The callback is optional and defaults to encoding, so third-party adapters keep their current behaviour unchanged.
+
+## Bug Fixes
+
+- fix: `Cache.ConCache.get_or_store/3` followed by `get/1` no longer raises. `get_or_store/3` writes through ConCache directly, bypassing the encode in `put/3`, so the matching `get/1` tried to `binary_to_term/1` a raw term.
+- fix: a binary value wrapped in braces but not valid JSON (eg `"{oops}"`) no longer raises `Jason.DecodeError` on read. `Cache.TermEncoder.decode/1` used `Jason.decode!/1`, and now falls back to returning the binary unchanged.
+- fix: raw `Cache.ETS` operations (`match_object/1`, `select/1`, `tab2list/0`, `foldl/2`) now see the terms that were `put`, rather than the opaque encoded binaries they used to return.
+
+## Breaking Changes
+
+- Values held by native-term adapters are now stored as terms rather than encoded binaries. This is not observable through `get/1`, `put/3` and `delete/1`, which round-trip exactly as before. It is observable if you read the underlying store directly (`:ets.lookup/2`, `:persistent_term.get/1`, `ConCache.get/2`) or through the raw ETS API — those now return terms, which is what they were always meant to return.
+- `Cache.DETS` is unchanged and still encodes, so existing `.dets` files stay readable. `Cache.ETS` with `:rehydration_path` also still encodes, so existing table dumps stay loadable.
+- An in-memory cache populated by an older version and read by this one would return raw binaries, but ETS, Agent, PersistentTerm and ConCache do not survive a restart, so there is no upgrade path on which that can happen.
+
 # 0.4.9
 
 ## Performance
diff --git a/guides/explanation/architecture.md b/guides/explanation/architecture.md
index 57d8157..84a50b4 100644
--- a/guides/explanation/architecture.md
+++ b/guides/explanation/architecture.md
@@ -37,11 +37,26 @@ When you `use Cache` in your module, the macro:
 
 ## Term Encoding and Compression
 
-ElixirCache includes internal term encoding functionality that handles serialization and deserialization of Elixir terms. This allows you to store complex Elixir data structures in any cache backend. The encoding system:
+ElixirCache lets you store any Elixir term in any backend. How a term gets there depends on
+what the backend can actually hold, and that is decided per adapter rather than globally.
 
-1. Uses Erlang's term_to_binary for efficient serialization
-2. Applies configurable compression to reduce memory usage
-3. Automatically handles decoding when retrieving values
+Adapters that store Erlang terms natively — `Cache.ETS`, `Cache.Agent`, `Cache.PersistentTerm`,
+`Cache.ConCache`, `Cache.Counter` — receive the term as-is. Serialising it first would cost a
+`term_to_binary/1` on every write and a `binary_to_term/1` on every read while buying nothing,
+and on a hot read path that decode dominates the lookup it is attached to.
+
+Adapters that cannot hold a term serialise it with `:erlang.term_to_binary/1`:
+
+- `Cache.Redis` stores bytes on the wire, so terms must be serialised before they leave the node.
+- `Cache.DETS` owns a durable on-disk format, and so does `Cache.ETS` when configured with
+  `:rehydration_path`. Both keep encoding so that files written by earlier versions stay readable.
+
+Setting `:compression_level` forces encoding on any adapter — an explicit request to compress
+is honoured even when the backend could have held the term directly.
+
+Adapters declare this through the optional `c:Cache.native_term_storage?/1` callback. It is
+resolved once at compile time, so there is no runtime branch on the read or write path.
+The callback is optional and defaults to encoding, so third-party adapters are unaffected.
 
 ## Sandboxing for Tests
 
diff --git a/lib/cache.ex b/lib/cache.ex
index 4c8de4f..5e71619 100644
--- a/lib/cache.ex
+++ b/lib/cache.ex
@@ -33,6 +33,22 @@ defmodule Cache do
               ErrorMessage.t_ok_res()
   @callback delete(cache_name :: atom, key :: atom | String.t()) :: ErrorMessage.t_ok_res()
 
+  @doc """
+  Returns true when the adapter stores Erlang terms natively, given its resolved options.
+
+  Adapters that hold terms in memory (`Cache.ETS`, `Cache.Agent`, `Cache.PersistentTerm`,
+  `Cache.ConCache`, `Cache.Counter`) do not need values to be run through
+  `:erlang.term_to_binary/1` first — doing so costs a full encode on every write and a
+  full decode on every read while buying nothing. Adapters that store bytes on the wire
+  (`Cache.Redis`) or that own a durable on-disk format (`Cache.DETS`) must return `false`.
+
+  This callback is optional. Adapters that do not implement it are assumed to need
+  encoding, which is what every adapter did before this callback existed.
+  """
+  @callback native_term_storage?(adapter_opts :: Keyword.t()) :: boolean()
+
+  @optional_callbacks native_term_storage?: 1
+
   defmacro __using__(opts) do
     sandbox_opt = Keyword.get(opts, :sandbox?, false)
 
@@ -288,6 +304,12 @@ defmodule Cache do
         @adapter_opts adapter_opts
         @compression_level if is_list(@adapter_opts), do: @adapter_opts[:compression_level]
 
+        # Resolved against the *configured* adapter and its *configured* opts rather than
+        # `@cache_adapter`/`@adapter_opts`, which are swapped out under `sandbox?: true`.
+        # A sandboxed cache must encode exactly like the adapter it stands in for, or a
+        # value would round-trip differently in test than in production.
+        @cache_encode_terms? Cache.TermEncoder.encoding_required?(opts[:adapter], opts[:opts])
+
         unquote(adapter_use_ast)
 
         def cache_name, do: @cache_name
@@ -334,8 +356,18 @@ defmodule Cache do
           @cache_adapter.child_spec({@cache_name, adapter_options()})
         end
 
+        @compile {:inline, encode_value: 1, decode_value: 1}
+
+        if @cache_encode_terms? do
+          defp encode_value(value), do: Cache.TermEncoder.encode(value, @compression_level)
+          defp decode_value(value), do: Cache.TermEncoder.decode(value)
+        else
+          defp encode_value(value), do: value
+          defp decode_value(value), do: value
+        end
+
         def put(key, ttl \\ nil, value) do
-          value = Cache.TermEncoder.encode(value, @compression_level)
+          value = encode_value(value)
           key = maybe_sandbox_key(key)
 
           :telemetry.span(
@@ -370,7 +402,7 @@ defmodule Cache do
                     res
 
                   {:ok, value} ->
-                    {:ok, Cache.TermEncoder.decode(value)}
+                    {:ok, decode_value(value)}
 
                   {:error, _} = error ->
                     error
diff --git a/lib/cache/agent.ex b/lib/cache/agent.ex
index 8757a27..2deeb8d 100644
--- a/lib/cache/agent.ex
+++ b/lib/cache/agent.ex
@@ -10,6 +10,12 @@ defmodule Cache.Agent do
   @impl Cache
   def opts_definition, do: []
 
+  @doc """
+  Values are held as terms in the Agent's map, so encoding them is pure overhead.
+  """
+  @impl Cache
+  def native_term_storage?(_opts), do: true
+
   @impl Cache
   def start_link(opts \\ []) do
     with {:error, {:already_started, pid}} <- Agent.start_link(fn -> %{} end, opts) do
diff --git a/lib/cache/con_cache.ex b/lib/cache/con_cache.ex
index f52e5bf..f498ffd 100644
--- a/lib/cache/con_cache.ex
+++ b/lib/cache/con_cache.ex
@@ -61,6 +61,15 @@ defmodule Cache.ConCache do
   @impl Cache
   def opts_definition, do: @opts_definition
 
+  @doc """
+  ConCache is ETS-backed and stores terms. Encoding is pure overhead, and it made
+  `get_or_store/3` unusable — that function writes through ConCache directly, bypassing
+  the encode in `Cache.put/3`, so a later `get/1` tried to `binary_to_term/1` a raw term
+  and raised.
+  """
+  @impl Cache
+  def native_term_storage?(_opts), do: true
+
   @impl Cache
   def start_link(opts) do
     cache_opts =
diff --git a/lib/cache/counter.ex b/lib/cache/counter.ex
index c62fb43..215644a 100644
--- a/lib/cache/counter.ex
+++ b/lib/cache/counter.ex
@@ -95,6 +95,13 @@ defmodule Cache.Counter do
   @impl Cache
   def opts_definition, do: @opts_definition
 
+  @doc """
+  Values are integers in an `:counters` array. `Cache.TermEncoder.encode/2` already
+  passed integers through untouched, so this only removes a no-op function call.
+  """
+  @impl Cache
+  def native_term_storage?(_opts), do: true
+
   @impl Cache
   def start_link(opts) do
     parent = self()
diff --git a/lib/cache/dets.ex b/lib/cache/dets.ex
index 3df2033..2c14284 100644
--- a/lib/cache/dets.ex
+++ b/lib/cache/dets.ex
@@ -363,6 +363,23 @@ defmodule Cache.DETS do
   @impl Cache
   def opts_definition, do: @opts_definition
 
+  @doc """
+  DETS keeps encoding, even though `:dets.insert/2` takes terms and would serialise
+  them itself.
+
+  The reason is the file on disk. Every `.dets` file written by an earlier version of
+  this library holds `:erlang.term_to_binary/1` blobs as its values. Those files outlive
+  the upgrade — that is the entire point of a disk cache. If this adapter switched to
+  native terms, the first read after an upgrade would hand the caller a raw binary where
+  it expects a term, silently, with no error to trace it back to.
+
+  The cost of keeping it is one redundant serialisation of data that is already headed
+  for a disk write, which is dominated by the I/O. The cost of dropping it is corrupting
+  every existing on-disk cache. That trade is not close.
+  """
+  @impl Cache
+  def native_term_storage?(_opts), do: false
+
   @impl Cache
   def start_link(opts) do
     parent = self()
diff --git a/lib/cache/ets.ex b/lib/cache/ets.ex
index f23169b..cfeae12 100644
--- a/lib/cache/ets.ex
+++ b/lib/cache/ets.ex
@@ -570,6 +570,20 @@ defmodule Cache.ETS do
   @impl Cache
   def opts_definition, do: @opts_definition
 
+  @doc """
+  `:ets.insert/2` stores terms, so encoding is pure overhead — and it actively breaks
+  the raw ETS API this adapter exposes (`match_object/1`, `select/1`, `tab2list/0`,
+  `foldl/2`), which sees opaque binaries instead of the terms that were put.
+
+  The exception is `:rehydration_path`. That option dumps the table to disk with
+  `:ets.tab2file/2` and reads it back on the next boot, so the table is a durable
+  format. Encoding is kept there for the same reason it is kept for `Cache.DETS`:
+  a table dumped by an older version holds `term_to_binary/1` blobs, and switching to
+  native terms would hand callers raw binaries where they expect terms.
+  """
+  @impl Cache
+  def native_term_storage?(opts), do: is_nil(opts[:rehydration_path])
+
   def opts_definition(opts) when is_list(opts) do
     if Keyword.keyword?(opts) do
       validate_and_normalize_ets_options(opts)
diff --git a/lib/cache/hash_ring.ex b/lib/cache/hash_ring.ex
index 53b6443..61ecf74 100644
--- a/lib/cache/hash_ring.ex
+++ b/lib/cache/hash_ring.ex
@@ -167,8 +167,8 @@ defmodule Cache.HashRing do
       {:ok, nil} ->
         read_repair(cache_name, key, target_node, underlying_adapter, underlying_opts, rpc)
 
-      {:ok, encoded} ->
-        {:ok, Cache.TermEncoder.decode(encoded)}
+      {:ok, stored} ->
+        {:ok, Cache.TermEncoder.maybe_decode(stored, underlying_adapter, underlying_opts)}
 
       {:error, _} = error ->
         error
@@ -180,7 +180,7 @@ defmodule Cache.HashRing do
     target_node = key_to_node(cache_name, key)
     rpc = adapter_opts[:rpc_module] || :erpc
     underlying_opts = validate_underlying_opts(underlying_adapter, Keyword.drop(adapter_opts, @strategy_keys))
-    encoded = Cache.TermEncoder.encode(value, underlying_opts[:compression_level])
+    encoded = Cache.TermEncoder.maybe_encode(value, underlying_adapter, underlying_opts)
 
     if target_node === Node.self() do
       underlying_adapter.put(cache_name, key, ttl, encoded, underlying_opts)
@@ -231,8 +231,8 @@ defmodule Cache.HashRing do
       end)
 
     case result do
-      {:found, encoded, old_node} ->
-        value = Cache.TermEncoder.decode(encoded)
+      {:found, stored, old_node} ->
+        value = Cache.TermEncoder.maybe_decode(stored, underlying_adapter, underlying_opts)
         migrate_value(cache_name, key, value, current_node, old_node, underlying_adapter, underlying_opts, rpc)
         {:ok, value}
 
@@ -242,7 +242,7 @@ defmodule Cache.HashRing do
   end
 
   defp migrate_value(cache_name, key, value, current_node, old_node, underlying_adapter, underlying_opts, rpc) do
-    encoded = Cache.TermEncoder.encode(value, underlying_opts[:compression_level])
+    encoded = Cache.TermEncoder.maybe_encode(value, underlying_adapter, underlying_opts)
 
     if current_node === Node.self() do
       underlying_adapter.put(cache_name, key, nil, encoded, underlying_opts)
diff --git a/lib/cache/persistent_term.ex b/lib/cache/persistent_term.ex
index 28d3085..66b60a5 100644
--- a/lib/cache/persistent_term.ex
+++ b/lib/cache/persistent_term.ex
@@ -27,6 +27,14 @@ defmodule Cache.PersistentTerm do
   @impl Cache
   def opts_definition, do: []
 
+  @doc """
+  `:persistent_term` exists to hand out zero-copy references to already-built terms.
+  Encoding a value into a binary and decoding it on every read defeats the entire
+  point of the adapter, so it never encodes.
+  """
+  @impl Cache
+  def native_term_storage?(_opts), do: true
+
   @impl Cache
   def start_link(opts) do
     Task.start_link(fn ->
diff --git a/lib/cache/redis.ex b/lib/cache/redis.ex
index 5e651b7..16f2929 100644
--- a/lib/cache/redis.ex
+++ b/lib/cache/redis.ex
@@ -200,6 +200,13 @@ defmodule Cache.Redis do
   @impl Cache
   def opts_definition, do: @opts_definition
 
+  @doc """
+  Redis stores bytes on the wire. Terms must be serialised before they leave the node,
+  so this adapter always encodes.
+  """
+  @impl Cache
+  def native_term_storage?(_opts), do: false
+
   @impl Cache
   def start_link(opts) do
     if !opts[:uri] do
diff --git a/lib/cache/refresh_ahead.ex b/lib/cache/refresh_ahead.ex
index e09298c..3cb0880 100644
--- a/lib/cache/refresh_ahead.ex
+++ b/lib/cache/refresh_ahead.ex
@@ -126,18 +126,16 @@ defmodule Cache.RefreshAhead do
         Keyword.drop(adapter_opts, [:refresh_before, :on_refresh, :lock_node_whitelist, :__cache_module__])
       )
 
-    compression_level = underlying_opts[:compression_level]
-
     case underlying_adapter.get(cache_name, key, underlying_opts) do
       {:ok, nil} ->
         {:ok, nil}
 
-      {:ok, encoded} ->
-        case Cache.TermEncoder.decode(encoded) do
+      {:ok, stored} ->
+        case Cache.TermEncoder.maybe_decode(stored, underlying_adapter, underlying_opts) do
           {value, inserted_at, ttl} ->
             maybe_refresh_async(
               cache_name, key, value, inserted_at, ttl,
-              underlying_adapter, adapter_opts, compression_level
+              underlying_adapter, adapter_opts, underlying_opts
             )
             {:ok, value}
 
@@ -159,9 +157,11 @@ defmodule Cache.RefreshAhead do
         Keyword.drop(adapter_opts, [:refresh_before, :on_refresh, :lock_node_whitelist, :__cache_module__])
       )
 
-    compression_level = underlying_opts[:compression_level]
     inserted_at = System.monotonic_time(:millisecond)
-    wrapped = Cache.TermEncoder.encode({value, inserted_at, ttl}, compression_level)
+
+    wrapped =
+      Cache.TermEncoder.maybe_encode({value, inserted_at, ttl}, underlying_adapter, underlying_opts)
+
     underlying_adapter.put(cache_name, key, ttl, wrapped, underlying_opts)
   end
 
@@ -179,21 +179,21 @@ defmodule Cache.RefreshAhead do
     underlying_adapter.delete(cache_name, key, underlying_opts)
   end
 
-  defp maybe_refresh_async(cache_name, key, _value, inserted_at, ttl, underlying_adapter, adapter_opts, compression_level)
+  defp maybe_refresh_async(cache_name, key, _value, inserted_at, ttl, underlying_adapter, adapter_opts, underlying_opts)
        when is_integer(ttl) do
     refresh_before = adapter_opts[:refresh_before]
     now = System.monotonic_time(:millisecond)
     age = now - inserted_at
 
     if age >= ttl - refresh_before do
-      maybe_spawn_refresh(cache_name, key, ttl, underlying_adapter, adapter_opts, compression_level)
+      maybe_spawn_refresh(cache_name, key, ttl, underlying_adapter, adapter_opts, underlying_opts)
     end
   end
 
-  defp maybe_refresh_async(_cache_name, _key, _value, _inserted_at, _ttl, _underlying_adapter, _adapter_opts, _compression_level),
+  defp maybe_refresh_async(_cache_name, _key, _value, _inserted_at, _ttl, _underlying_adapter, _adapter_opts, _underlying_opts),
     do: :ok
 
-  defp maybe_spawn_refresh(cache_name, key, ttl, underlying_adapter, adapter_opts, compression_level) do
+  defp maybe_spawn_refresh(cache_name, key, ttl, underlying_adapter, adapter_opts, underlying_opts) do
     tracker = tracker_name(cache_name)
 
     if safe_ets_insert_new(tracker, {key, true}) do
@@ -209,16 +209,15 @@ defmodule Cache.RefreshAhead do
 
             case invoke_refresh(on_refresh, cache_module, key) do
               {:ok, new_value} ->
-                underlying_opts =
-                  Keyword.drop(adapter_opts, [
-                    :refresh_before,
-                    :on_refresh,
-                    :lock_node_whitelist,
-                    :__cache_module__
-                  ])
-
                 new_inserted_at = System.monotonic_time(:millisecond)
-                wrapped = Cache.TermEncoder.encode({new_value, new_inserted_at, ttl}, compression_level)
+
+                wrapped =
+                  Cache.TermEncoder.maybe_encode(
+                    {new_value, new_inserted_at, ttl},
+                    underlying_adapter,
+                    underlying_opts
+                  )
+
                 underlying_adapter.put(cache_name, key, ttl, wrapped, underlying_opts)
 
               {:error, _} ->
diff --git a/lib/cache/sandbox.ex b/lib/cache/sandbox.ex
index dba1524..c72404e 100644
--- a/lib/cache/sandbox.ex
+++ b/lib/cache/sandbox.ex
@@ -55,6 +55,14 @@ defmodule Cache.Sandbox do
   @impl Cache
   def opts_definition, do: []
 
+  @doc """
+  The sandbox holds terms in an Agent map. It is never consulted directly to decide
+  whether to encode — `Cache` resolves that against the adapter the sandbox stands in
+  for, so a sandboxed cache round-trips values exactly like the real one.
+  """
+  @impl Cache
+  def native_term_storage?(_opts), do: true
+
   @impl Cache
   def start_link(opts \\ []) do
     with {:error, {:already_started, pid}} <- Agent.start_link(fn -> %{} end, opts) do
diff --git a/lib/cache/term_encoder.ex b/lib/cache/term_encoder.ex
index 84cabe7..0cf9a27 100644
--- a/lib/cache/term_encoder.ex
+++ b/lib/cache/term_encoder.ex
@@ -1,6 +1,117 @@
 defmodule Cache.TermEncoder do
   @moduledoc false
 
+  # SECTION ADAPTER CAPABILITY
+  #
+  # Adapters that store Erlang terms natively (ETS, Agent, PersistentTerm, ConCache,
+  # Counter) do not need `term_to_binary/1` — the round trip is pure overhead and it
+  # is the dominant cost on a hot read path. Adapters that store bytes (Redis) or
+  # that own a durable on-disk format (DETS) still need it.
+  #
+  # Resolution is done once at compile time by the `Cache` macro, so there is no
+  # runtime branch on the read/write path.
+
+  @doc """
+  Returns true when values must be encoded to a binary before being handed to the adapter.
+
+  Encoding is required when:
+
+    * the adapter stores bytes rather than terms (`Cache.Redis`), or
+    * the adapter persists an encoded format to disk (`Cache.DETS`,
+      `Cache.ETS` with `:rehydration_path`), or
+    * `:compression_level` is set — the caller explicitly asked for compression, or
+    * the adapter opts are resolved at runtime, so the capability cannot be read at
+      compile time. Encoding is the safe default: it is what every adapter did before.
+
+  Third-party adapters that do not implement `c:Cache.native_term_storage?/1` keep
+  encoding, so they are unaffected.
+  """
+  @spec encoding_required?(module() | {module(), term()}, keyword() | term()) :: boolean()
+  def encoding_required?(adapter, adapter_opts)
+
+  def encoding_required?(adapter, nil), do: encoding_required?(adapter, [])
+
+  def encoding_required?(_adapter, adapter_opts) when not is_list(adapter_opts), do: true
+
+  def encoding_required?(adapter, adapter_opts) do
+    if is_nil(adapter_opts[:compression_level]) do
+      not native_term_storage?(adapter, adapter_opts)
+    else
+      true
+    end
+  end
+
+  @doc """
+  Returns true when the adapter stores Erlang terms natively, given its options.
+
+  Strategy adapters (`{Cache.HashRing, Cache.ETS}`) delegate to the adapter they wrap.
+  """
+  @spec native_term_storage?(module() | {module(), term()}, keyword()) :: boolean()
+  def native_term_storage?(adapter, adapter_opts)
+
+  def native_term_storage?({_strategy_module, underlying_adapter}, adapter_opts)
+      when is_atom(underlying_adapter) do
+    native_term_storage?(underlying_adapter, adapter_opts)
+  end
+
+  # Strategies configured with something other than an adapter module (eg the layer
+  # list of `Cache.MultiLayer`) only reach this path under `sandbox?: true`, where the
+  # store is `Cache.Sandbox` and holds terms natively.
+  def native_term_storage?({_strategy_module, _strategy_config}, _adapter_opts), do: true
+
+  def native_term_storage?(adapter, adapter_opts) when is_atom(adapter) do
+    if function_exported?(adapter, :native_term_storage?, 1) do
+      adapter.native_term_storage?(adapter_opts)
+    else
+      compile_time_native_term_storage?(adapter, adapter_opts)
+    end
+  end
+
+  # `function_exported?/3` is false for a module that has not been loaded yet, which
+  # happens when an adapter is compiled in the same project as the cache using it.
+  # `Code.ensure_compiled/1` blocks on the parallel compiler so the answer is stable
+  # across builds rather than depending on compilation order.
+  defp compile_time_native_term_storage?(adapter, adapter_opts) do
+    case Code.ensure_compiled(adapter) do
+      {:module, module} ->
+        function_exported?(module, :native_term_storage?, 1) and
+          module.native_term_storage?(adapter_opts)
+
+      {:error, _reason} ->
+        false
+    end
+  end
+
+  @doc """
+  Encodes only when the adapter needs bytes. Used by strategy adapters, which resolve
+  their underlying adapter at runtime rather than at compile time.
+  """
+  @spec maybe_encode(term(), module(), keyword()) :: term()
+  def maybe_encode(term, adapter, adapter_opts) do
+    if encoding_required?(adapter, adapter_opts) do
+      encode(term, compression_level(adapter_opts))
+    else
+      term
+    end
+  end
+
+  @doc """
+  Decodes only when the adapter needs bytes. The inverse of `maybe_encode/3`.
+  """
+  @spec maybe_decode(term(), module(), keyword()) :: term()
+  def maybe_decode(term, adapter, adapter_opts) do
+    if encoding_required?(adapter, adapter_opts) do
+      decode(term)
+    else
+      term
+    end
+  end
+
+  defp compression_level(adapter_opts) when is_list(adapter_opts), do: adapter_opts[:compression_level]
+  defp compression_level(_adapter_opts), do: nil
+
+  # SECTION ENCODING
+
   def encode(term, compression_level)
       when not is_nil(compression_level) and compression_level >= 1 do
     :erlang.term_to_binary(term,
@@ -30,7 +141,7 @@ defmodule Cache.TermEncoder do
         String.to_integer(binary)
 
       binary =~ ~r/^{.*}$/ ->
-        Jason.decode!(binary)
+        decode_json(binary)
 
       true ->
         binary
diff --git a/mix.exs b/mix.exs
index 6f638d0..b00dc5b 100644
--- a/mix.exs
+++ b/mix.exs
@@ -4,7 +4,7 @@ defmodule ElixirCache.MixProject do
   def project do
     [
       app: :elixir_cache,
-      version: "0.4.9",
+      version: "0.5.0",
       elixir: "~> 1.11",
       start_permanent: Mix.env() == :prod,
       description:
diff --git a/test/cache/hash_ring_test.exs b/test/cache/hash_ring_test.exs
index b687d2a..d5bc6f3 100644
--- a/test/cache/hash_ring_test.exs
+++ b/test/cache/hash_ring_test.exs
@@ -158,11 +158,12 @@ defmodule Cache.HashRingTest do
     test "recovers value via read-repair from old ring owner" do
       cache_name = :test_hash_ring_cache
       fake_node = :fake_live@node
-      encoded = Cache.TermEncoder.encode("repaired_value", nil)
+      # Cache.ETS stores terms natively, so a real remote node returns the term as-is.
+      stored = "repaired_value"
 
       rpc_module = build_rpc(fn node, _mod, func, _args ->
         cond do
-          node === fake_node and func === :get -> {:ok, encoded}
+          node === fake_node and func === :get -> {:ok, stored}
           func === :put -> :ok
           func === :delete -> :ok
           true -> {:ok, nil}
@@ -180,7 +181,7 @@ defmodule Cache.HashRingTest do
     test "migrates value to current node and schedules async delete on repair" do
       cache_name = :test_hash_ring_cache
       fake_node = :fake_migrate@node
-      encoded = Cache.TermEncoder.encode("migrate_me", nil)
+      stored = "migrate_me"
       calls_agent = :"repair_calls_#{:erlang.unique_integer([:positive])}"
       {:ok, _} = Agent.start_link(fn -> [] end, name: calls_agent)
 
@@ -188,7 +189,7 @@ defmodule Cache.HashRingTest do
         Agent.update(calls_agent, fn acc -> [{node, func} | acc] end)
 
         cond do
-          node === fake_node and func === :get -> {:ok, encoded}
+          node === fake_node and func === :get -> {:ok, stored}
           func === :put -> :ok
           func === :delete -> :ok
           true -> {:ok, nil}
diff --git a/test/cache/multi_layer_test.exs b/test/cache/multi_layer_test.exs
index b6bc470..3fa55ba 100644
--- a/test/cache/multi_layer_test.exs
+++ b/test/cache/multi_layer_test.exs
@@ -38,6 +38,20 @@ defmodule Cache.MultiLayerTest do
       opts: [backfill_ttl: 5000]
   end
 
+  defmodule MixedRedisLayer do
+    use Cache,
+      adapter: Cache.Redis,
+      name: :ml_redis_layer,
+      opts: [uri: "redis://localhost:6379"]
+  end
+
+  defmodule MixedLayerCache do
+    use Cache,
+      adapter: {Cache.MultiLayer, [FastCache, MixedRedisLayer]},
+      name: :mixed_layer_cache,
+      opts: []
+  end
+
   setup do
     start_supervised!(%{
       id: :fast_cache_sup,
@@ -156,4 +170,50 @@ defmodule Cache.MultiLayerTest do
       assert Cache.Strategy.strategy?(Cache.MultiLayer) === true
     end
   end
+
+  describe "encoding across layers of different adapters" do
+    setup do
+      start_supervised!({Cache, [MixedRedisLayer]})
+
+      start_supervised!(%{
+        id: :mixed_layer_sup,
+        type: :supervisor,
+        start: {Cache, :start_link, [[MixedLayerCache], [name: :mixed_layer_sup]]}
+      })
+
+      # Cleaned up front rather than in on_exit, which runs after the pool is torn down.
+      Cache.Redis.command!(:ml_redis_layer, ["DEL", "ml_redis_layer:mixed_key"])
+
+      :ok
+    end
+
+    test "each layer stores in its own native format — ETS gets the term, Redis gets bytes" do
+      value = %{user: "mika", roles: [:admin]}
+
+      assert :ok === MixedLayerCache.put("mixed_key", value)
+
+      # The ETS layer holds the term itself — no encode/decode on the hot read path.
+      assert [{"mixed_key", ^value}] = :ets.lookup(:ml_fast_cache, "mixed_key")
+
+      # The Redis layer holds an encoded binary, because bytes are all Redis can hold.
+      raw = Cache.Redis.command!(:ml_redis_layer, ["GET", "ml_redis_layer:mixed_key"])
+
+      assert is_binary(raw)
+      assert :erlang.binary_to_term(raw) === value
+
+      assert {:ok, ^value} = MixedLayerCache.get("mixed_key")
+    end
+
+    test "a value backfilled from the Redis layer lands in ETS as a term" do
+      value = %{backfilled: true}
+
+      assert :ok === MixedRedisLayer.put("mixed_key", value)
+      assert {:ok, nil} === FastCache.get("mixed_key")
+
+      # Reading through the layers hits Redis, decodes, and backfills ETS.
+      assert {:ok, ^value} = MixedLayerCache.get("mixed_key")
+
+      assert [{"mixed_key", ^value}] = :ets.lookup(:ml_fast_cache, "mixed_key")
+    end
+  end
 end
diff --git a/test/cache/native_term_storage_test.exs b/test/cache/native_term_storage_test.exs
new file mode 100644
index 0000000..108aa88
--- /dev/null
+++ b/test/cache/native_term_storage_test.exs
@@ -0,0 +1,227 @@
+defmodule Cache.NativeTermStorageTest do
+  @moduledoc """
+  Proves values are encoded only for adapters that actually need bytes.
+
+  Adapters that store Erlang terms natively (ETS, Agent, PersistentTerm, ConCache)
+  must hand the term straight to the store, and adapters that store bytes (Redis) or
+  own a durable on-disk format (DETS, ETS with `:rehydration_path`) must keep encoding.
+  """
+
+  use ExUnit.Case, async: true
+
+  defmodule ETSCache do
+    use Cache, adapter: Cache.ETS, name: :nts_ets, opts: []
+  end
+
+  defmodule AgentCache do
+    use Cache, adapter: Cache.Agent, name: :nts_agent, opts: []
+  end
+
+  defmodule PersistentTermCache do
+    use Cache, adapter: Cache.PersistentTerm, name: :nts_persistent_term, opts: []
+  end
+
+  defmodule ConCacheCache do
+    use Cache, adapter: Cache.ConCache, name: :nts_con_cache, opts: []
+  end
+
+  defmodule RedisCache do
+    use Cache, adapter: Cache.Redis, name: :nts_redis, opts: [uri: "redis://localhost:6379"]
+  end
+
+  defmodule DETSCache do
+    use Cache, adapter: Cache.DETS, name: :nts_dets, opts: [file_path: "/tmp/nts_dets"]
+  end
+
+  defstruct [:name, :count]
+
+  @native_caches [ETSCache, AgentCache, PersistentTermCache, ConCacheCache]
+
+  setup do
+    start_supervised!({Cache, [ETSCache, AgentCache, PersistentTermCache, ConCacheCache, RedisCache, DETSCache]})
+
+    :ok
+  end
+
+  describe "&Cache.TermEncoder.encoding_required?/2" do
+    test "is false for adapters that store terms natively" do
+      refute Cache.TermEncoder.encoding_required?(Cache.ETS, [])
+      refute Cache.TermEncoder.encoding_required?(Cache.Agent, [])
+      refute Cache.TermEncoder.encoding_required?(Cache.PersistentTerm, [])
+      refute Cache.TermEncoder.encoding_required?(Cache.ConCache, [])
+      refute Cache.TermEncoder.encoding_required?(Cache.Counter, [])
+    end
+
+    test "is true for adapters that store bytes or persist to disk" do
+      assert Cache.TermEncoder.encoding_required?(Cache.Redis, [])
+      assert Cache.TermEncoder.encoding_required?(Cache.DETS, [])
+    end
+
+    test "is true for a native adapter that persists its table to disk" do
+      assert Cache.TermEncoder.encoding_required?(Cache.ETS, rehydration_path: "/tmp/nts")
+    end
+
+    test "is true when compression_level is set — the caller asked for compression" do
+      assert Cache.TermEncoder.encoding_required?(Cache.ETS, compression_level: 6)
+      assert Cache.TermEncoder.encoding_required?(Cache.Agent, compression_level: 1)
+    end
+
+    test "defaults to true for a third-party adapter that does not implement the callback" do
+      defmodule ThirdPartyAdapter do
+        @behaviour Cache
+
+        @impl Cache
+        def opts_definition, do: []
+        @impl Cache
+        def start_link(_opts), do: :ignore
+        @impl Cache
+        def child_spec({_name, _opts}), do: %{id: __MODULE__, start: {__MODULE__, :start_link, [[]]}}
+        @impl Cache
+        def get(_name, _key, _opts \\ []), do: {:ok, nil}
+        @impl Cache
+        def put(_name, _key, _ttl \\ nil, _value, _opts \\ []), do: :ok
+        @impl Cache
+        def delete(_name, _key, _opts \\ []), do: :ok
+      end
+
+      assert Cache.TermEncoder.encoding_required?(ThirdPartyAdapter, [])
+    end
+
+    test "is true when opts are resolved at runtime and cannot be read at compile time" do
+      assert Cache.TermEncoder.encoding_required?(Cache.ETS, {:my_app, :some_key})
+      assert Cache.TermEncoder.encoding_required?(Cache.ETS, {MyMod, :opts, []})
+    end
+
+    test "a strategy delegates to the adapter it wraps" do
+      refute Cache.TermEncoder.encoding_required?({Cache.HashRing, Cache.ETS}, [])
+      assert Cache.TermEncoder.encoding_required?({Cache.RefreshAhead, Cache.Redis}, [])
+    end
+  end
+
+  describe "native adapters store the raw term" do
+    test "the value in the underlying store is the term itself, not a binary" do
+      value = %{user: "mika", roles: [:admin]}
+
+      for cache <- @native_caches do
+        assert :ok === cache.put(:raw_key, value)
+      end
+
+      assert [{:raw_key, ^value}] = :ets.lookup(:nts_ets, :raw_key)
+      assert %{raw_key: ^value} = Agent.get(:nts_agent, & &1)
+      assert ^value = :persistent_term.get({:nts_persistent_term, :raw_key})
+      assert ^value = ConCache.get(:nts_con_cache, :raw_key)
+    end
+
+    test "arbitrary terms round-trip identically without encoding" do
+      terms = [
+        %{nested: %{list: [1, 2, 3]}},
+        {:tuple, "with", 3, :parts},
+        %__MODULE__{name: "struct", count: 7},
+        [keyword: :list, another: 1],
+        self(),
+        :an_atom,
+        "a plain string",
+        123,
+        1.5,
+        nil,
+        <<1, 2, 3>>
+      ]
+
+      for cache <- @native_caches, term <- terms do
+        assert :ok === cache.put(:round_trip, term)
+        assert {:ok, ^term} = cache.get(:round_trip)
+      end
+    end
+
+    test "a pid round-trips as the same live pid, not a decoded copy" do
+      assert :ok === ETSCache.put(:pid_key, self())
+      assert {:ok, pid} = ETSCache.get(:pid_key)
+
+      assert pid === self()
+      assert Process.alive?(pid)
+    end
+
+    test "raw ETS operations see real terms, so match_object and tab2list work" do
+      assert :ok === ETSCache.put(:mika, %{name: "mika"})
+
+      assert [mika: %{name: "mika"}] === ETSCache.tab2list()
+      assert [mika: %{name: "mika"}] === ETSCache.match_object({:_, %{name: "mika"}})
+    end
+
+    test "a ConCache get_or_store writes a value that get can read back" do
+      assert "stored" === ConCacheCache.get_or_store(:gos_key, nil, fn -> "stored" end)
+      assert {:ok, "stored"} === ConCacheCache.get(:gos_key)
+    end
+
+    test "a PersistentTerm read hands out the same term on every read, with no copy" do
+      value = Enum.to_list(1..500)
+
+      assert :ok === PersistentTermCache.put(:zero_copy, value)
+      assert {:ok, first} = PersistentTermCache.get(:zero_copy)
+      assert {:ok, second} = PersistentTermCache.get(:zero_copy)
+
+      assert :erts_debug.same(first, second)
+    end
+  end
+
+  describe "adapters that need bytes still encode" do
+    test "a Redis value is stored as an encoded binary and decoded on read" do
+      value = %{user: "mika", roles: [:admin]}
+
+      assert :ok === RedisCache.put("encoded_key", value)
+
+      raw = Cache.Redis.command!(:nts_redis, ["GET", "nts_redis:encoded_key"])
+
+      assert is_binary(raw)
+      assert :erlang.binary_to_term(raw) === value
+      assert {:ok, ^value} = RedisCache.get("encoded_key")
+    end
+
+    test "a DETS value is stored as an encoded binary so existing on-disk files stay readable" do
+      on_exit(fn -> File.rm_rf("/tmp/nts_dets") end)
+
+      value = %{user: "mika"}
+
+      assert :ok === DETSCache.put(:dets_key, value)
+      assert [{:dets_key, raw}] = :dets.lookup(:nts_dets, :dets_key)
+
+      assert is_binary(raw)
+      assert :erlang.binary_to_term(raw) === value
+      assert {:ok, ^value} = DETSCache.get(:dets_key)
+    end
+
+    test "a DETS file written by an older version still decodes to the original term" do
+      on_exit(fn -> File.rm_rf("/tmp/nts_dets") end)
+
+      legacy = :erlang.term_to_binary(%{written_by: "0.4.9"})
+      :dets.insert(:nts_dets, {:legacy_key, legacy})
+
+      assert {:ok, %{written_by: "0.4.9"}} === DETSCache.get(:legacy_key)
+    end
+
+    test "an explicit compression_level still compresses on a native adapter" do
+      value = String.duplicate("compress me ", 500)
+      opts = [compression_level: 6]
+
+      encoded = Cache.TermEncoder.maybe_encode(value, Cache.ETS, opts)
+
+      assert is_binary(encoded)
+      assert byte_size(encoded) < byte_size(:erlang.term_to_binary(value))
+      assert Cache.TermEncoder.maybe_decode(encoded, Cache.ETS, opts) === value
+    end
+  end
+
+  describe "binaries that look like JSON" do
+    test "a brace-wrapped string round-trips unchanged on a native adapter" do
+      assert :ok === ETSCache.put(:json_ish, ~s({"a": 1}))
+
+      assert {:ok, ~s({"a": 1})} === ETSCache.get(:json_ish)
+    end
+
+    test "a brace-wrapped string that is not valid JSON no longer raises on read" do
+      assert :ok === RedisCache.put("not_json", "{oops not json}")
+
+      assert {:ok, "{oops not json}"} === RedisCache.get("not_json")
+    end
+  end
+end
diff --git a/test/cache/refresh_ahead_test.exs b/test/cache/refresh_ahead_test.exs
index 9a6f8be..123b23d 100644
--- a/test/cache/refresh_ahead_test.exs
+++ b/test/cache/refresh_ahead_test.exs
@@ -155,6 +155,13 @@ defmodule Cache.RefreshAheadTest do
       Process.sleep(250)
 
       assert {:ok, "original"} === LockedRefreshCache.get("locked_key")
+
+      # That get spawned a refresh task. Wait for it to lose the race for the lock and
+      # clean itself up before releasing — otherwise it can reach :global.set_lock after
+      # the del_lock below, take the freed lock, and refresh the value we assert is still
+      # untouched.
+      Process.sleep(250)
+
       assert true === :global.del_lock(lock_id, lock_nodes)
 
       assert {:ok, "original"} === LockedRefreshCache.get("locked_key")