diff --git a/CHANGELOG.md b/CHANGELOG.md
index 33f3807..d044dd3 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -19,6 +19,17 @@ its own schema and says exactly what it is. `Latu.Result.Arrow` is the reader 
 streaming format, no dependency — and `Latu.Result.Nx` the mapping, behind the now-optional
 `:nx`. Adding `{:nx, "~> 0.13"}` is what turns them on; without it `to_nx/2` says so.
 
+**Every RPC retries on the session's `Latu.Retry`**, not only the result stream — PySpark's
+own behaviour — except the best-effort releases. A `RetryInfo` on an error makes it retryable
+whatever its status, with its delay as a floor under the backoff, capped by the new
+`max_server_retry_delay` (10 min); `%Latu.Error{}` gains `retry_delay`. A unary call's
+`[:latu, :retry, :attempt]` event carries `rpc` where an execution's carries `operation_id`.
+Smaller: a cleartext token is allowed to every loopback address, not four spellings of it, and
+an IPv6 literal host connects at all (`sc://[::1]:15002` crashed inside elixir-grpc);
+`SPARK_USER` precedes the OS user as the default `user_id`; `lit/1` refuses a non-UTF-8 binary
+and names Spark's `X'…'` literal; `true`/`false` are refused where a column name is taken
+instead of naming a column `"true"`. No migration.
+
 **`Latu.error_details/2` restores the whole message.** The server abbreviates a gRPC status
 message to 2048 characters, so a long analysis error arrived cut short with `...`; the detail
 carries it whole, and the call now puts it on the error beside the causes. No migration.
diff --git a/docs/decisions.md b/docs/decisions.md
index 996c59a..cb41f41 100644
--- a/docs/decisions.md
+++ b/docs/decisions.md
@@ -1339,3 +1339,41 @@ gRPC status cut to 2048 characters ending in `...`, and that is what `%Latu.Erro
 the same on the explicit call: `error_details/2` puts the thrown error's message on the struct
 beside the causes. Nothing changes without the call, by the same reasoning that makes the fetch
 explicit; a short message is identical on both paths.
+\n
+## 2026-09-07 — Every RPC retries, `RetryInfo` counts, and four smaller alignments
+
+**Unary RPCs retry on the session's policy.** PySpark wraps every call in `Retrying`; Latu
+retried only the execution stream, so a transient `UNAVAILABLE` on `AnalyzePlan` or `Config`
+failed outright. `Client.retrying/3` now wraps every call that asks the server for something.
+Not `ExecutePlan`/`ReattachExecute` — `Latu.Client.Execution` owns those — and not the releases
+that are best effort: `ReleaseExecute` on the stream, and `ReleaseSession` inside
+`disconnect(release: true)`, which has a five-second budget to keep. Backing off blocks the
+caller's process, as the reattach path already does. The retry event carries `rpc` where an
+execution's carries `operation_id`.
+
+**A `RetryInfo` makes any status retryable, and its delay is a floor.** PySpark's
+`DefaultPolicy` third case; `Latu.Retry.retryable?/1` is now the one predicate, with
+`max_server_retry_delay` (10 min) capping the floor before jitter. Spark's own server never
+attaches one — `ErrorUtils`/`SparkConnectService` on 4.2 have no `RetryInfo` — so this is for
+a gateway in front of it. The old comment blaming elixir-grpc for not decoding the detail was
+wrong: `details` is the same `Any` list `ErrorInfo` is read from.
+
+**Loopback is what the kernel keeps on the machine**: `localhost`, 127/8, `::1` in any
+spelling, and 127/8 as an IPv4-mapped address — parsed with `:inet.parse_address/1` rather than
+listed. PySpark checks the string `"localhost"` alone. Testing it found that an IPv6 literal
+never connected: `"host:port"` compatibility mode splits on colons, so the target takes
+elixir-grpc's `ipv6:[addr]:port` form when the host parses as one.
+
+**`SPARK_USER` comes before the OS user** for the default `user_id`, PySpark's own order. The
+`user-agent` header stays unset: PySpark sends its agent only as `client_type`, so there was no
+parity to gain.
+
+**A binary literal must be UTF-8**: the proto field is a string and the server refuses invalid
+bytes with an opaque error; `lit/1` refuses first and names Spark's `X'…'` literal. **Booleans
+are refused wherever a name is taken**: they are atoms, and `to_name(true)` was a column called
+`"true"`.
+
+Not done: caching the `create_dataframe/3` config probe on the returned session. A caller
+rarely threads `df.session` back, so the saving would seldom land, and a cached threshold goes
+stale the moment `set_conf/3` moves it — SparkEx shipped exactly that bug. One round trip per
+call stays.
diff --git a/lib/latu.ex b/lib/latu.ex
index ea1ed61..863c4fe 100644
--- a/lib/latu.ex
+++ b/lib/latu.ex
@@ -80,7 +80,7 @@ defmodule Latu do
     * `:keepalive` — HTTP/2 ping interval in milliseconds. Defaults to `60_000`.
     * `:keepalive_tolerance` — pings missed before the channel is considered dead. Defaults
       to `2`.
-    * `:retry` — a `Latu.Retry` policy every execution retries under. Defaults to `%Latu.Retry{}`.
+    * `:retry` — a `Latu.Retry` policy every RPC retries under. Defaults to `%Latu.Retry{}`.
 
   Why these numbers, and why the knobs live on the session rather than in application config,
   is in `docs/decisions.md`.
@@ -779,7 +779,9 @@ defmodule Latu do
   name. When no data name is in the schema, the schema renames by position —
   `create_dataframe(session, [n: [1]], schema: "id INT")` — and a schema naming some columns but
   not others is refused, naming them. The server parses the schema as the frame is built (one
-  round trip), so a malformed one fails here rather than at the first action.
+  round trip), so a malformed one fails here rather than at the first action. Every column
+  ships nullable — Explorer writes Arrow fields so — and the server refuses to cast one to a
+  `NOT NULL` field (`NULLABLE_COLUMN_OR_FIELD`); leave nullability to the default.
 
   ## Examples
 
diff --git a/lib/latu/client.ex b/lib/latu/client.ex
index 4673ed9..d3305af 100644
--- a/lib/latu/client.ex
+++ b/lib/latu/client.ex
@@ -12,11 +12,10 @@ defmodule Latu.Client do
   alias Latu.Progress
   alias Latu.Protocol.Spark.Connect, as: Proto
   alias Latu.Protocol.Spark.Connect.SparkConnectService.Stub
+  alias Latu.Retry
   alias Latu.Session
   alias Latu.Telemetry
 
-  @loopback ~w(localhost 127.0.0.1 ::1 0:0:0:0:0:0:0:1)
-
   # The one field each AnalyzePlan response arm carries. Read out of PySpark's
   # `AnalyzeResult.fromProto`; `persist` and `unpersist` carry none.
   @analyze_fields %{
@@ -94,7 +93,7 @@ defmodule Latu.Client do
   end
 
   defp released(%Session{} = session) do
-    case release_session(session, timeout: @release_on_disconnect_timeout) do
+    case release_session(session, [timeout: @release_on_disconnect_timeout], &rpc/3) do
       {:ok, session} ->
         session
 
@@ -163,7 +162,7 @@ defmodule Latu.Client do
 
     call = fn -> Stub.analyze_plan(session.channel, request, timeout: session.timeout) end
 
-    with {:ok, response} <- rpc("AnalyzePlan", session, call),
+    with {:ok, response} <- retrying("AnalyzePlan", session, call),
          {:ok, session} <-
            Session.confirm(session, response.session_id, response.server_side_session_id) do
       {:ok, response, session}
@@ -305,7 +304,7 @@ defmodule Latu.Client do
 
     call = fn -> Stub.config(session.channel, request, timeout: session.timeout) end
 
-    with {:ok, response} <- rpc("Config", session, call),
+    with {:ok, response} <- retrying("Config", session, call),
          {:ok, session} <-
            Session.confirm(session, response.session_id, response.server_side_session_id) do
       warn(op, response.warnings)
@@ -400,7 +399,7 @@ defmodule Latu.Client do
     request = struct!(request, interrupt_scope(opts))
     call = fn -> Stub.interrupt(session.channel, request, timeout: session.timeout) end
 
-    with {:ok, response} <- rpc("Interrupt", session, call),
+    with {:ok, response} <- retrying("Interrupt", session, call),
          {:ok, session} <-
            Session.confirm(session, response.session_id, response.server_side_session_id) do
       {:ok, response.interrupted_ids, session}
@@ -444,7 +443,8 @@ defmodule Latu.Client do
 
     call = fn -> Stub.get_status(session.channel, request, timeout: session.timeout) end
 
-    with {:ok, response} <- rpc("GetStatus", session, call) |> explain_missing_session(session),
+    with {:ok, response} <-
+           retrying("GetStatus", session, call) |> explain_missing_session(session),
          {:ok, session} <-
            Session.confirm(session, response.session_id, response.server_side_session_id) do
       {:ok, Enum.map(response.operation_statuses, &operation_status/1), session}
@@ -512,7 +512,7 @@ defmodule Latu.Client do
     call = fn -> Stub.clone_session(session.channel, request, timeout: session.timeout) end
 
     with {:ok, response} <-
-           rpc("CloneSession", session, call) |> explain_missing_session(session),
+           retrying("CloneSession", session, call) |> explain_missing_session(session),
          {:ok, session} <-
            Session.confirm(session, response.session_id, response.server_side_session_id),
          {:ok, clone} <- cloned(session, response, wanted) do
@@ -553,6 +553,12 @@ defmodule Latu.Client do
   end
 
   def release_session(%Session{} = session, opts) when is_list(opts) do
+    release_session(session, opts, &retrying/3)
+  end
+
+  # `through` is `retrying/3` for the call a caller makes, `rpc/3` for the courtesy on the way
+  # out of `disconnect/2`, which has a budget of its own to keep.
+  defp release_session(%Session{} = session, opts, through) do
     opts = Keyword.validate!(opts, allow_reconnect: false, timeout: session.timeout)
 
     request = %Proto.ReleaseSessionRequest{
@@ -564,7 +570,7 @@ defmodule Latu.Client do
 
     call = fn -> Stub.release_session(session.channel, request, timeout: opts[:timeout]) end
 
-    with {:ok, response} <- rpc("ReleaseSession", session, call) do
+    with {:ok, response} <- through.("ReleaseSession", session, call) do
       Session.confirm(session, response.session_id, response.server_side_session_id)
     end
   end
@@ -614,7 +620,7 @@ defmodule Latu.Client do
 
     call = fn -> Stub.artifact_status(session.channel, request, timeout: session.timeout) end
 
-    with {:ok, response} <- rpc("ArtifactStatus", session, call),
+    with {:ok, response} <- retrying("ArtifactStatus", session, call),
          {:ok, session} <-
            Session.confirm(session, response.session_id, response.server_side_session_id) do
       cached =
@@ -642,7 +648,7 @@ defmodule Latu.Client do
       |> GRPC.Stub.recv(timeout: session.timeout)
     end
 
-    with {:ok, response} <- rpc("AddArtifacts", session, call),
+    with {:ok, response} <- retrying("AddArtifacts", session, call),
          {:ok, session} <-
            Session.confirm(session, response.session_id, response.server_side_session_id) do
       case Enum.reject(response.artifacts, & &1.is_crc_successful) do
@@ -1040,6 +1046,35 @@ defmodule Latu.Client do
     %Proto.UserContext{user_id: session.user_id, user_name: session.user_name}
   end
 
+  @doc false
+  # Every RPC that asks the server for something is retried on the session's policy, as
+  # PySpark's `Retrying` wraps every call. Not the two that open a result stream —
+  # `Latu.Client.Execution` decides those — and not a best-effort release, which must never
+  # hold a caller up. Backing off blocks the caller's process, the one place Latu has to do it.
+  # Public for the offline test; `call` is any zero-arity function returning what a `Stub` call
+  # does.
+  def retrying(name, %Session{retry: %Retry{} = retry} = session, call) do
+    retrying(name, session, call, retry, 0)
+  end
+
+  defp retrying(name, session, call, retry, attempt) do
+    case rpc(name, session, call) do
+      {:error, %Error{} = error} = failure ->
+        if attempt < retry.max_retries and Retry.retryable?(error) do
+          backoff = Retry.wait(retry, attempt, error.retry_delay)
+          ids = %{session_id: session.session_id, rpc: name}
+          Telemetry.attempt(:retry, backoff, attempt + 1, ids)
+          Process.sleep(backoff)
+          retrying(name, session, call, retry, attempt + 1)
+        else
+          failure
+        end
+
+      result ->
+        result
+    end
+  end
+
   # Every unary RPC funnels through here, which is why the telemetry span is here and not at
   # ten call sites. Only the session *id* is handed over — see `Latu.Telemetry`.
   defp rpc(name, %Session{} = session, call) do
@@ -1062,16 +1097,24 @@ defmodule Latu.Client do
   # `deps/googleapis`, which `grpc_core` depends on — so this needs no dependency, no proto
   # generation and no round trip.
   @error_info_url "type.googleapis.com/google.rpc.ErrorInfo"
+  @retry_info_url "type.googleapis.com/google.rpc.RetryInfo"
 
   defp rpc_error(%GRPC.RPCError{} = error) do
     Error.new(:rpc, error.message, [status: error.status, details: error.details] ++ info(error))
   end
 
   defp info(%GRPC.RPCError{details: details}) when is_list(details) do
-    case Enum.find(details, &(&1.type_url == @error_info_url)) do
-      nil -> []
-      any -> from_error_info(Google.Rpc.ErrorInfo.decode(any.value))
-    end
+    Enum.flat_map(details, fn
+      %{type_url: @error_info_url, value: value} ->
+        from_error_info(Google.Rpc.ErrorInfo.decode(value))
+
+      # Spark itself never attaches one; a gateway in front of it may, to say "later".
+      %{type_url: @retry_info_url, value: value} ->
+        [retry_delay: delay_ms(Google.Rpc.RetryInfo.decode(value))]
+
+      _other ->
+        []
+    end)
   rescue
     # Inspecting an error must never be the thing that fails. A trailer Latu cannot read leaves
     # the fields nil, which is exactly what a server that sent none does.
@@ -1080,6 +1123,13 @@ defmodule Latu.Client do
 
   defp info(%GRPC.RPCError{}), do: []
 
+  # An absent duration is still a RetryInfo — "retry, no particular wait" — so 0, not nil.
+  defp delay_ms(%Google.Rpc.RetryInfo{retry_delay: nil}), do: 0
+
+  defp delay_ms(%Google.Rpc.RetryInfo{retry_delay: %{seconds: seconds, nanos: nanos}}) do
+    seconds * 1_000 + div(nanos, 1_000_000)
+  end
+
   # The metadata keys are PySpark's — `connect.py`'s `convert_exception` reads the same six.
   defp from_error_info(%Google.Rpc.ErrorInfo{metadata: metadata}) do
     [
@@ -1133,7 +1183,7 @@ defmodule Latu.Client do
 
     call = fn -> Stub.fetch_error_details(session.channel, request, timeout: session.timeout) end
 
-    with {:ok, response} <- rpc("FetchErrorDetails", session, call),
+    with {:ok, response} <- retrying("FetchErrorDetails", session, call),
          {:ok, _session} <-
            Session.confirm(session, response.session_id, response.server_side_session_id) do
       {:ok, filled(error, Enum.map(causes(response), &cause/1))}
@@ -1207,8 +1257,14 @@ defmodule Latu.Client do
   # "host:port" puts elixir-grpc in compatibility mode, which rewrites to `ipv4:host:port`
   # and resolves through Gun rather than the DNS resolver. That is what we want: a Spark
   # Connect session is server-side state pinned to one server, so re-resolving or balancing
-  # across A records would silently move us to a server that has never heard of it.
-  defp target(%Session{host: host, port: port}), do: "#{host}:#{port}"
+  # across A records would silently move us to a server that has never heard of it. An IPv6
+  # literal has colons of its own, so it takes the one form elixir-grpc reads brackets in.
+  defp target(%Session{host: host, port: port}) do
+    case :inet.parse_ipv6strict_address(String.to_charlist(host)) do
+      {:ok, _address} -> "ipv6:[#{host}]:#{port}"
+      {:error, _not_ipv6} -> "#{host}:#{port}"
+    end
+  end
 
   defp open(target, opts) do
     case GRPC.Stub.connect(target, opts) do
@@ -1287,13 +1343,28 @@ defmodule Latu.Client do
   defp check_token_transport(%Session{token: nil}), do: :ok
   defp check_token_transport(%Session{use_ssl: true}), do: :ok
 
-  defp check_token_transport(%Session{host: host}) when host in @loopback, do: :ok
-
   defp check_token_transport(%Session{host: host}) do
-    {:error,
-     Error.new(
-       :connect,
-       "refusing to send a bearer token in cleartext to #{host}; add ;use_ssl=true to the URL"
-     )}
+    if loopback?(host) do
+      :ok
+    else
+      {:error,
+       Error.new(
+         :connect,
+         "refusing to send a bearer token in cleartext to #{host}; add ;use_ssl=true to the URL"
+       )}
+    end
+  end
+
+  # What never leaves the machine: `localhost`, all of 127/8, `::1` in any spelling, and 127/8
+  # written as an IPv4-mapped IPv6 address. PySpark checks the string "localhost" alone.
+  defp loopback?("localhost"), do: true
+
+  defp loopback?(host) do
+    case :inet.parse_address(String.to_charlist(host)) do
+      {:ok, {127, _, _, _}} -> true
+      {:ok, {0, 0, 0, 0, 0, 0, 0, 1}} -> true
+      {:ok, {0, 0, 0, 0, 0, 0xFFFF, high, _low}} -> div(high, 256) == 127
+      _ -> false
+    end
   end
 end
diff --git a/lib/latu/client/execution.ex b/lib/latu/client/execution.ex
index d5b458a..6dd5ad0 100644
--- a/lib/latu/client/execution.ex
+++ b/lib/latu/client/execution.ex
@@ -18,11 +18,6 @@ defmodule Latu.Client.Execution do
   # The retry policy proper is a separate concern. See dev/README.md.
   @max_empty_reattaches 100
 
-  # gRPC status codes, spelled out because this module must not reference GRPC — see the
-  # layering test.
-  @unavailable 14
-  @internal 13
-
   # Spark reports a lost execution or session in the message, with no status code of its own to
   # match on. PySpark detects it the same way.
   @lost_handle ["INVALID_HANDLE.OPERATION_NOT_FOUND", "INVALID_HANDLE.SESSION_NOT_FOUND"]
@@ -143,15 +138,17 @@ defmodule Latu.Client.Execution do
       # The server has no record of this execution. If nothing had arrived yet, the original
       # ExecutePlan never landed and re-sending it under the same operation_id is safe.
       lost_handle?(error) and is_nil(execution.last_response_id) and budget ->
-        {{:restart, Retry.wait(retry, retries)}, %{execution | retries: retries + 1}}
+        wait = Retry.wait(retry, retries, error.retry_delay)
+        {{:restart, wait}, %{execution | retries: retries + 1}}
 
       # If responses had already arrived, they are gone and re-sending would duplicate the ones
       # that did. PySpark raises RESPONSE_ALREADY_RECEIVED here for the same reason.
       lost_handle?(error) ->
         {{:fail, unrecoverable(execution, error)}, execution}
 
-      budget and retryable?(error) ->
-        {{:reattach, Retry.wait(retry, retries)}, %{execution | retries: retries + 1}}
+      budget and Retry.retryable?(error) ->
+        wait = Retry.wait(retry, retries, error.retry_delay)
+        {{:reattach, wait}, %{execution | retries: retries + 1}}
 
       true ->
         {{:fail, give_up(execution, error)}, execution}
@@ -162,23 +159,14 @@ defmodule Latu.Client.Execution do
   # Retries
   # =============================================
 
-  # Which errors are retryable is PySpark's DefaultPolicy; how often and how long is the
-  # session's `Latu.Retry`. PySpark's third case — any error carrying RetryInfo metadata — is
-  # not covered: elixir-grpc does not decode that detail.
+  # Which errors are retryable, how often and how long is the session's `Latu.Retry`; the one
+  # case of its own here is a lost handle, which only an execution can have.
   defp policy(%__MODULE__{session: %Session{retry: %Retry{} = retry}}), do: retry
 
   defp lost_handle?(%Error{message: message}) do
     Enum.any?(@lost_handle, &String.contains?(message, &1))
   end
 
-  defp retryable?(%Error{status: @unavailable}), do: true
-
-  defp retryable?(%Error{status: @internal, message: message}) do
-    message =~ "INVALID_CURSOR.DISCONNECTED"
-  end
-
-  defp retryable?(%Error{}), do: false
-
   # Nothing had arrived, so the only way here is a spent budget.
   defp unrecoverable(%__MODULE__{last_response_id: nil} = execution, %Error{} = error) do
     give_up(execution, error)
diff --git a/lib/latu/data_frame.ex b/lib/latu/data_frame.ex
index 07cafd8..750d3c7 100644
--- a/lib/latu/data_frame.ex
+++ b/lib/latu/data_frame.ex
@@ -353,7 +353,10 @@ defmodule Latu.DataFrame do
   end
 
   defp column_name(name) when is_binary(name), do: name
-  defp column_name(name) when is_atom(name) and not is_nil(name), do: Atom.to_string(name)
+
+  defp column_name(name) when is_atom(name) and not is_nil(name) and not is_boolean(name) do
+    Atom.to_string(name)
+  end
 
   defp column_name(name) do
     raise ArgumentError, "a column name is a string or an atom, not #{inspect(name)}"
diff --git a/lib/latu/error.ex b/lib/latu/error.ex
index 5b6c134..5ecb4e6 100644
--- a/lib/latu/error.ex
+++ b/lib/latu/error.ex
@@ -29,6 +29,8 @@ defmodule Latu.Error do
       send one. **Not** in `message/1`: a JVM trace is not what you want in a REPL, and it is
       one field away when you do.
     * `error_id` — the handle `Latu.error_details/2` fetches the full cause chain with.
+    * `retry_delay` — milliseconds the server asked the client to wait before trying again,
+      when it attached a `RetryInfo`; `nil` otherwise. Anything carrying one is retried.
     * `causes` — only populated by `Latu.error_details/2`. One entry per exception in the
       chain, root cause last. The same call restores a `message` the server abbreviated to
       2048 characters on the wire.
@@ -46,6 +48,7 @@ defmodule Latu.Error do
     :sql_state,
     :stacktrace,
     :error_id,
+    :retry_delay,
     classes: [],
     parameters: %{},
     causes: []
@@ -68,6 +71,7 @@ defmodule Latu.Error do
           sql_state: String.t() | nil,
           stacktrace: String.t() | nil,
           error_id: String.t() | nil,
+          retry_delay: non_neg_integer() | nil,
           classes: [String.t()],
           parameters: %{optional(String.t()) => String.t()},
           causes: [cause()]
diff --git a/lib/latu/plan.ex b/lib/latu/plan.ex
index bdd9a4f..17988cf 100644
--- a/lib/latu/plan.ex
+++ b/lib/latu/plan.ex
@@ -274,12 +274,18 @@ defmodule Latu.Plan do
   defp path(path), do: raise(ArgumentError, "a path is a string, not #{inspect(path)}")
 
   defp table_name(name) when is_binary(name), do: name
-  defp table_name(name) when is_atom(name) and not is_nil(name), do: Atom.to_string(name)
+
+  defp table_name(name) when is_atom(name) and not is_nil(name) and not is_boolean(name) do
+    Atom.to_string(name)
+  end
 
   defp table_name(name), do: name_of(name, "table name")
 
   defp name_of(name, _what) when is_binary(name), do: name
-  defp name_of(name, _what) when is_atom(name) and not is_nil(name), do: Atom.to_string(name)
+
+  defp name_of(name, _what) when is_atom(name) and not is_nil(name) and not is_boolean(name) do
+    Atom.to_string(name)
+  end
 
   defp name_of(name, what) do
     raise ArgumentError, "a #{what} is a string or an atom, not #{inspect(name)}"
@@ -1204,7 +1210,7 @@ defmodule Latu.Plan do
   defp identifier(name, what \\ "column name")
   defp identifier(name, _what) when is_binary(name), do: name
 
-  defp identifier(name, _what) when is_atom(name) and not is_nil(name) do
+  defp identifier(name, _what) when is_atom(name) and not is_nil(name) and not is_boolean(name) do
     Atom.to_string(name)
   end
 
@@ -1991,7 +1997,19 @@ defmodule Latu.Plan do
   defp scalar(value) when is_integer(value) and value in @int32, do: {:integer, value}
   defp scalar(value) when is_integer(value) and value in @int64, do: {:long, value}
   defp scalar(value) when is_float(value), do: {:double, value}
-  defp scalar(value) when is_binary(value), do: {:string, value}
+
+  # Elixir's binary is both Spark's string and its bytes; the proto field is a string, and the
+  # server refuses one that is not UTF-8. Spark's own spelling for bytes is the SQL literal.
+  defp scalar(value) when is_binary(value) do
+    if String.valid?(value) do
+      {:string, value}
+    else
+      raise ArgumentError,
+            "a binary literal is a UTF-8 string; for raw bytes use Spark's binary literal, " <>
+              "expr(\"X'…'\") with the bytes from Base.encode16/1"
+    end
+  end
+
   defp scalar(%Date{} = value), do: {:date, Date.diff(value, @epoch_date)}
 
   defp scalar(%DateTime{} = value), do: {:timestamp, DateTime.to_unix(value, :microsecond)}
@@ -2113,7 +2131,8 @@ defmodule Latu.Plan do
   def to_projections(columns) when is_list(columns), do: Enum.map(columns, &to_projection/1)
   def to_projections(column), do: to_projections([column])
 
-  defp to_projection({name, value}) when is_atom(name) and not is_nil(name) do
+  defp to_projection({name, value})
+       when is_atom(name) and not is_nil(name) and not is_boolean(name) do
     as(to_expr(value), name)
   end
 
@@ -2441,7 +2460,10 @@ defmodule Latu.Plan do
   # A statistic is neither a column name nor a value, so it gets its own coercion and its own
   # message: "25%" is a perfectly good one and would read oddly as a column.
   defp statistic(name) when is_binary(name), do: name
-  defp statistic(name) when is_atom(name) and not is_nil(name), do: Atom.to_string(name)
+
+  defp statistic(name) when is_atom(name) and not is_nil(name) and not is_boolean(name) do
+    Atom.to_string(name)
+  end
 
   defp statistic(other) do
     raise ArgumentError,
diff --git a/lib/latu/retry.ex b/lib/latu/retry.ex
index 70b1209..ab48f2a 100644
--- a/lib/latu/retry.ex
+++ b/lib/latu/retry.ex
@@ -14,25 +14,42 @@ defmodule Latu.Retry do
   Times are in milliseconds. `max_retries: 0` turns retrying off, which is what a test that
   wants a failure to surface immediately wants.
 
-  What is retried is not configurable, and deliberately: an `UNAVAILABLE`, a disconnected
-  cursor, and a lost handle with nothing received yet.
+  What is retried is not configurable, and deliberately — `retryable?/1` is PySpark's list: an
+  `UNAVAILABLE`, a disconnected cursor, anything the server attached a `RetryInfo` to, and, for
+  an execution, a lost handle with nothing received yet. A `RetryInfo`'s delay is a floor under
+  the backoff, capped at `max_server_retry_delay`.
   """
 
+  alias Latu.Error
+
   @default_max_retries 15
   @default_initial_backoff 50
   @default_max_backoff 60_000
   @default_backoff_multiplier 4.0
   @default_jitter 500
   @default_min_jitter_threshold 2_000
+  @default_max_server_retry_delay 600_000
+
+  @counts [
+    :max_retries,
+    :initial_backoff,
+    :max_backoff,
+    :jitter,
+    :min_jitter_threshold,
+    :max_server_retry_delay
+  ]
 
-  @counts [:max_retries, :initial_backoff, :max_backoff, :jitter, :min_jitter_threshold]
+  # gRPC status codes, spelled out: nothing outside `Latu.Client` may reference GRPC.
+  @unavailable 14
+  @internal 13
 
   defstruct max_retries: @default_max_retries,
             initial_backoff: @default_initial_backoff,
             max_backoff: @default_max_backoff,
             backoff_multiplier: @default_backoff_multiplier,
             jitter: @default_jitter,
-            min_jitter_threshold: @default_min_jitter_threshold
+            min_jitter_threshold: @default_min_jitter_threshold,
+            max_server_retry_delay: @default_max_server_retry_delay
 
   @type t :: %__MODULE__{
           max_retries: non_neg_integer(),
@@ -40,7 +57,8 @@ defmodule Latu.Retry do
           max_backoff: non_neg_integer(),
           backoff_multiplier: number(),
           jitter: non_neg_integer(),
-          min_jitter_threshold: non_neg_integer()
+          min_jitter_threshold: non_neg_integer(),
+          max_server_retry_delay: non_neg_integer()
         }
 
   @doc """
@@ -62,22 +80,42 @@ defmodule Latu.Retry do
     raise ArgumentError, "retry is a keyword list or a %Latu.Retry{}, not #{inspect(other)}"
   end
 
+  @doc """
+  Whether a failed call is worth trying again — PySpark's `DefaultPolicy.can_retry`.
+
+  `UNAVAILABLE`; `INTERNAL` naming a disconnected cursor; or any status at all when the server
+  attached a `RetryInfo`, which is how a gateway says "later" rather than "no".
+  """
+  @spec retryable?(Error.t()) :: boolean()
+  def retryable?(%Error{retry_delay: delay}) when is_integer(delay), do: true
+  def retryable?(%Error{status: @unavailable}), do: true
+
+  def retryable?(%Error{status: @internal, message: message}) do
+    message =~ "INVALID_CURSOR.DISCONNECTED"
+  end
+
+  def retryable?(%Error{}), do: false
+
   @doc """
   How long to wait before attempt `attempt`, counting from zero.
 
       iex> Latu.Retry.wait(Latu.Retry.new(), 0)
       50
 
-  Jitter above `min_jitter_threshold` is the only nondeterminism in the transport, and it is
-  bounded by `jitter`.
+  A `floor` — the server's `RetryInfo` delay, when it sent one — lifts the wait to at least
+  that, capped at `max_server_retry_delay`; jitter goes on after. Jitter above
+  `min_jitter_threshold` is the only nondeterminism in the transport, and it is bounded by
+  `jitter`.
   """
-  @spec wait(t(), non_neg_integer()) :: non_neg_integer()
-  def wait(%__MODULE__{} = retry, attempt) when is_integer(attempt) and attempt >= 0 do
+  @spec wait(t(), non_neg_integer(), non_neg_integer() | nil) :: non_neg_integer()
+  def wait(%__MODULE__{} = retry, attempt, floor \\ nil)
+      when is_integer(attempt) and attempt >= 0 do
     capped = min(retry.initial_backoff * retry.backoff_multiplier ** attempt, retry.max_backoff)
+    lifted = max(capped, min(floor || 0, retry.max_server_retry_delay))
 
-    if capped > retry.min_jitter_threshold,
-      do: trunc(capped + :rand.uniform() * retry.jitter),
-      else: trunc(capped)
+    if lifted > retry.min_jitter_threshold,
+      do: trunc(lifted + :rand.uniform() * retry.jitter),
+      else: trunc(lifted)
   end
 
   defp validated!(%__MODULE__{} = retry) do
diff --git a/lib/latu/session.ex b/lib/latu/session.ex
index 3b99e90..8ab9a55 100644
--- a/lib/latu/session.ex
+++ b/lib/latu/session.ex
@@ -11,7 +11,7 @@ defmodule Latu.Session do
 
     * `:timeout`, `:connect_timeout` — per-RPC and establishment deadlines.
     * `:window_size`, `:keepalive`, `:keepalive_tolerance` — HTTP/2 flow control and liveness.
-    * `:retry` — a `Latu.Retry`, the policy every execution retries under.
+    * `:retry` — a `Latu.Retry`, the policy every RPC retries under.
 
   Every one is a `Latu.connect/2` option, so none of them needs a struct poke.
 
@@ -368,7 +368,10 @@ defmodule Latu.Session do
   # Defaults
   # =============================================
 
-  defp default_user_id, do: System.get_env("USER") || System.get_env("USERNAME") || ""
+  # PySpark's own order: SPARK_USER, then the OS user.
+  defp default_user_id do
+    System.get_env("SPARK_USER") || System.get_env("USER") || System.get_env("USERNAME") || ""
+  end
 
   defp default_client_type do
     "latu/#{@version} elixir/#{System.version()} otp/#{System.otp_release()}"
diff --git a/lib/latu/telemetry.ex b/lib/latu/telemetry.ex
index 9ef8e4d..c774933 100644
--- a/lib/latu/telemetry.ex
+++ b/lib/latu/telemetry.ex
@@ -20,8 +20,9 @@ defmodule Latu.Telemetry do
 
   Metadata is `rpc`, `outcome` and `error_class` on the RPC events, and `session_id` plus
   `operation_id` on everything that belongs to an execution. The RPC events carry no
-  `operation_id`; `[:latu, :execute, :stop]` carries an `outcome` of `:ok`, `:error` or
-  `:abandoned`. Durations are in native time units, as `:telemetry.span/3`'s are.
+  `operation_id`; a `[:latu, :retry, :attempt]` for a unary call carries `rpc` in its place;
+  `[:latu, :execute, :stop]` carries an `outcome` of `:ok`, `:error` or `:abandoned`.
+  Durations are in native time units, as `:telemetry.span/3`'s are.
 
   **`[:latu, :rpc, :*]` covers every gRPC call, including `ExecutePlan` — but for that one and
   `ReattachExecute` it measures *opening* the stream, not draining it**, because a Latu result
diff --git a/test/integration/connection_test.exs b/test/integration/connection_test.exs
index 1adc823..453f2bf 100644
--- a/test/integration/connection_test.exs
+++ b/test/integration/connection_test.exs
@@ -66,6 +66,18 @@ defmodule Latu.Integration.ConnectionTest do
     refute match?({:error, %Error{message: "refusing" <> _}}, result)
   end
 
+  test "in every spelling the kernel keeps on the machine" do
+    for host <- ["127.0.0.2", "[::1]", "[0:0:0:0:0:0:0:1]", "[::ffff:127.0.0.1]"] do
+      result = Latu.connect("sc://#{host}:1/;token=s3cr3t", connect_timeout: 500)
+      refute match?({:error, %Error{message: "refusing" <> _}}, result), host
+    end
+  end
+
+  test "but not to a host that only looks local" do
+    assert {:error, %Error{message: "refusing" <> _}} =
+             Latu.connect("sc://127.0.0.1.example.com:1/;token=s3cr3t", connect_timeout: 500)
+  end
+
   test "spark_version/1 reports the server's version" do
     session = Latu.connect!(@url)
     assert {:ok, version} = Latu.spark_version(session)
diff --git a/test/latu/client/execution_test.exs b/test/latu/client/execution_test.exs
index 0709e24..226b31d 100644
--- a/test/latu/client/execution_test.exs
+++ b/test/latu/client/execution_test.exs
@@ -237,6 +237,15 @@ defmodule Latu.Client.ExecutionTest do
       assert {{:fail, _}, _} = Execution.step(execution, no_status)
     end
 
+    test "unless it carries a RetryInfo, whose delay is then the floor", %{
+      execution: execution
+    } do
+      throttled = Error.new(:rpc, "RESOURCE_EXHAUSTED: later", status: 8, retry_delay: 3_000)
+
+      assert {{:reattach, wait}, _} = Execution.step(execution, {:error, throttled})
+      assert wait >= 3_000 and wait < 3_500
+    end
+
     test "back off on PySpark's schedule", %{execution: execution} do
       {delays, _} =
         Enum.map_reduce(1..7, execution, fn _, execution ->
diff --git a/test/latu/client_test.exs b/test/latu/client_test.exs
index 4bd566e..acb00cb 100644
--- a/test/latu/client_test.exs
+++ b/test/latu/client_test.exs
@@ -15,5 +15,64 @@ defmodule Latu.ClientTest do
     assert message =~ "not connected"
   end
 
+  describe "retrying/3" do
+    setup do
+      retry = [initial_backoff: 1, jitter: 0, max_retries: 3]
+      %{session: Session.from_url!("sc://h", retry: retry)}
+    end
+
+    test "tries a retryable failure again, on the session's schedule", %{session: session} do
+      {calls, answer} = flaky(2, {:ok, :answer})
+
+      assert Client.retrying("Test", session, answer) == {:ok, :answer}
+      assert Agent.get(calls, & &1) == 3
+    end
+
+    test "hands the failure back once the budget is spent", %{session: session} do
+      {calls, never} = flaky(99, {:ok, :never})
+
+      assert {:error, %Error{status: 14}} = Client.retrying("Test", session, never)
+      assert Agent.get(calls, & &1) == 4
+    end
+
+    test "does not try again what is not worth it", %{session: session} do
+      refused = fn -> {:error, %GRPC.RPCError{status: 3, message: "INVALID_ARGUMENT"}} end
+
+      assert {:error, %Error{status: 3}} = Client.retrying("Test", session, refused)
+    end
+
+    test "each attempt is a retry event naming the RPC", %{session: session} do
+      # A remote capture, as telemetry_test does: `:telemetry.attach/4` logs about a local
+      # one. The handler is global, so it forwards this session's events alone.
+      handler = "retrying-#{inspect(make_ref())}"
+      config = {self(), session.session_id}
+      :telemetry.attach(handler, [:latu, :retry, :attempt], &__MODULE__.forward/4, config)
+      on_exit(fn -> :telemetry.detach(handler) end)
+
+      {_calls, answer} = flaky(1, {:ok, :answer})
+      assert {:ok, :answer} = Client.retrying("Test", session, answer)
+
+      assert_received {%{attempt: 1, backoff: 1}, %{rpc: "Test"}}
+    end
+
+    def forward(_event, measurements, metadata, {test, session_id}) do
+      if metadata[:session_id] == session_id, do: send(test, {measurements, metadata})
+    end
+
+    # A call that fails `failures` times with UNAVAILABLE, then answers — returning what a
+    # `Stub` call returns, since `rpc/3` is what turns that into a `%Latu.Error{}`.
+    defp flaky(failures, then) do
+      {:ok, calls} = Agent.start_link(fn -> 0 end)
+
+      call = fn ->
+        if Agent.get_and_update(calls, &{&1 + 1, &1 + 1}) <= failures,
+          do: {:error, %GRPC.RPCError{status: 14, message: "UNAVAILABLE: connection reset"}},
+          else: then
+      end
+
+      {calls, call}
+    end
+  end
+
   defp plan, do: Plan.new(Plan.range(0, 5, 1))
 end
diff --git a/test/latu/plan_test.exs b/test/latu/plan_test.exs
index faa9938..dea4b59 100644
--- a/test/latu/plan_test.exs
+++ b/test/latu/plan_test.exs
@@ -286,6 +286,16 @@ defmodule Latu.PlanTest do
       assert arm(Plan.to_expr(true)) == {:boolean, true}
     end
 
+    test "and in a name position they are refused, not columns named true" do
+      for bad <- [true, false, nil] do
+        assert_raise ArgumentError, ~r/string or an atom/, fn -> apply(Plan, :to_name, [bad]) end
+      end
+    end
+
+    test "a binary literal is UTF-8; raw bytes have Spark's own spelling" do
+      assert_raise ArgumentError, ~r/UTF-8.*X'/, fn -> apply(Plan, :lit, [<<0xFF, 0xFE>>]) end
+    end
+
     test "an expression passes through untouched" do
       expression = Plan.expr("id > 3")
 
diff --git a/test/latu/retry_test.exs b/test/latu/retry_test.exs
index 92e063c..0b9f000 100644
--- a/test/latu/retry_test.exs
+++ b/test/latu/retry_test.exs
@@ -1,6 +1,7 @@
 defmodule Latu.RetryTest do
   use ExUnit.Case, async: true
 
+  alias Latu.Error
   alias Latu.Retry
 
   doctest Latu.Retry
@@ -13,7 +14,8 @@ defmodule Latu.RetryTest do
                max_backoff: 60_000,
                backoff_multiplier: 4.0,
                jitter: 500,
-               min_jitter_threshold: 2_000
+               min_jitter_threshold: 2_000,
+               max_server_retry_delay: 600_000
              } = Retry.new()
     end
 
@@ -81,5 +83,27 @@ defmodule Latu.RetryTest do
 
       assert Enum.map(0..3, &Retry.wait(retry, &1)) == [10, 20, 40, 80]
     end
+
+    test "a server's delay is a floor under the backoff, capped at max_server_retry_delay" do
+      retry = Retry.new(jitter: 0)
+
+      assert Retry.wait(retry, 0, 3_000) == 3_000
+      assert Retry.wait(retry, 99, 3_000) == 60_000
+      assert Retry.wait(retry, 0, 3_600_000) == 600_000
+      assert Retry.wait(retry, 0, 0) == 50
+      assert Retry.wait(retry, 0, nil) == 50
+    end
+  end
+
+  describe "retryable?/1" do
+    test "UNAVAILABLE, a disconnected cursor, and anything carrying a RetryInfo" do
+      assert Retry.retryable?(Error.new(:rpc, "UNAVAILABLE", status: 14))
+      assert Retry.retryable?(Error.new(:rpc, "INVALID_CURSOR.DISCONNECTED", status: 13))
+      assert Retry.retryable?(Error.new(:rpc, "RESOURCE_EXHAUSTED", status: 8, retry_delay: 0))
+
+      refute Retry.retryable?(Error.new(:rpc, "INTERNAL: bad plan", status: 13))
+      refute Retry.retryable?(Error.new(:rpc, "RESOURCE_EXHAUSTED", status: 8))
+      refute Retry.retryable?(Error.new(:decode, "not the server's"))
+    end
   end
 end
diff --git a/test/latu/session_test.exs b/test/latu/session_test.exs
index 085f1ff..ac2ecde 100644
--- a/test/latu/session_test.exs
+++ b/test/latu/session_test.exs
@@ -186,6 +186,20 @@ defmodule Latu.SessionTest do
                Session.from_url("sc://h", connect_timeout: 500)
     end
 
+    test "the default user id is SPARK_USER before the OS user, PySpark's order" do
+      previous = System.get_env("SPARK_USER")
+      System.put_env("SPARK_USER", "svc-latu")
+
+      on_exit(fn ->
+        if previous,
+          do: System.put_env("SPARK_USER", previous),
+          else: System.delete_env("SPARK_USER")
+      end)
+
+      assert {:ok, %Session{user_id: "svc-latu"}} = Session.from_url("sc://h")
+      assert {:ok, %Session{user_id: "alice"}} = Session.from_url("sc://h/;user_id=alice")
+    end
+
     test "user_name is empty unless asked for, matching PySpark's UserContext" do
       assert {:ok, %Session{user_name: ""}} = Session.from_url("sc://h/;user_id=alice")
       assert {:ok, %Session{user_name: "Alice"}} = Session.from_url("sc://h", user_name: "Alice")