fix(sdk): 0.3.4 must-fixes — encode module loading + raw response bodies (+5 hardening commits)

.github/workflows/ci.yml

@@ -32,7 +32,7 @@ jobs:
         uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 # v4.2.2
         with:
           repository: ArchAstro/archastro-openapi
-          ref: ${{ github.event_name == 'pull_request' && 'features/calvin-archastro-06-08-2026-elixir-sdk' || 'main' }}
+          ref: main
           path: .ci/archastro-openapi
 
       - name: Setup Node (for prism + @archastro/channel-harness)

lib/archastro/channel.ex

@@ -21,8 +21,13 @@ defmodule ArchAstro.SDK.Channel do
           ArchAstro.SDK.Codec.descriptor(),
           timeout()
         ) :: {:ok, t()} | {:error, ArchAstro.SDK.Error.reason()}
-  def join(socket, topic, payload, module, descriptor, timeout \\ 5_000),
-    do: GenServer.call(socket, {:archastro_join, topic, payload, module, descriptor}, timeout)
+  # Payloads are encoded here rather than in the socket process: an
+  # unencodable value must fail its own caller, not take down the shared
+  # socket and every other channel riding on it.
+  def join(socket, topic, payload, module, descriptor, timeout \\ 5_000) do
+    encoded = ArchAstro.SDK.Codec.encode(payload)
+    GenServer.call(socket, {:archastro_join, topic, encoded, module, descriptor}, timeout)
+  end
 
   @spec push(
           t(),
@@ -31,13 +36,15 @@ defmodule ArchAstro.SDK.Channel do
           ArchAstro.SDK.Codec.descriptor(),
           timeout()
         ) :: {:ok, ArchAstro.SDK.JSON.t() | struct() | :ok} | {:error, ArchAstro.SDK.Error.reason()}
-  def push(%__MODULE__{} = channel, event, payload, descriptor, timeout \\ 5_000),
-    do:
-      GenServer.call(
-        channel.socket,
-        {:archastro_push, channel.topic, event, payload, descriptor},
-        timeout
-      )
+  def push(%__MODULE__{} = channel, event, payload, descriptor, timeout \\ 5_000) do
+    encoded = ArchAstro.SDK.Codec.encode(payload)
+
+    GenServer.call(
+      channel.socket,
+      {:archastro_push, channel.topic, event, encoded, descriptor},
+      timeout
+    )
+  end
 
   @spec subscribe(t(), String.t(), pid(), ArchAstro.SDK.Codec.descriptor()) :: :ok
   def subscribe(%__MODULE__{} = channel, event, subscriber, descriptor)

lib/archastro/client.ex

@@ -43,7 +43,7 @@ defmodule ArchAstro.SDK.Client do
            opts
            |> Keyword.get(:base_url, ArchAstro.SDK.GeneratedClient.default_base_url())
            |> String.trim_trailing("/"),
-         request: Keyword.get(opts, :req, Req.new(retry: false)),
+         request: Keyword.get(opts, :req, Req.new(retry: false, redirect: false)),
          token_binding: binding,
          telemetry_metadata: Keyword.get(opts, :telemetry_metadata, %{})
        }}

lib/archastro/codec.ex

@@ -73,8 +73,21 @@ defmodule ArchAstro.SDK.Codec do
   def encode(%DateTime{} = value), do: DateTime.to_iso8601(value)
   def encode(%Date{} = value), do: Date.to_iso8601(value)
 
+  def encode(%NaiveDateTime{} = value) do
+    raise ArgumentError,
+          "cannot encode NaiveDateTime #{inspect(value)}: the API requires an " <>
+            "ISO 8601 datetime with a UTC offset; convert it with " <>
+            "DateTime.from_naive!(value, \"Etc/UTC\") first"
+  end
+
+  def encode(%Time{} = value) do
+    raise ArgumentError,
+          "cannot encode Time #{inspect(value)}: the API has no time-of-day wire " <>
+            "format; send a full DateTime instead"
+  end
+
   def encode(%module{} = value) do
-    if function_exported?(module, :to_map, 1),
+    if Code.ensure_loaded?(module) and function_exported?(module, :to_map, 1),
       do: module.to_map(value),
       else: Map.from_struct(value)
   end

lib/archastro/http.ex

@@ -38,6 +38,11 @@ defmodule ArchAstro.SDK.HTTP do
       retry: false
     ]
 
+    request_opts =
+      if Keyword.get(opts, :raw, false),
+        do: Keyword.put(request_opts, :decode_body, false),
+        else: request_opts
+
     request_opts =
       case Keyword.fetch(opts, :body) do
         {:ok, value} -> Keyword.put(request_opts, :json, Codec.encode(value))
@@ -70,7 +75,7 @@ defmodule ArchAstro.SDK.HTTP do
         {:ok, decoded}
 
       {:ok, response} ->
-        error = Error.from_response(response)
+        error = response |> decode_error_body() |> Error.from_response()
 
         :telemetry.execute(
           [:archastro, :request, :stop],
@@ -92,4 +97,15 @@ defmodule ArchAstro.SDK.HTTP do
         {:error, error}
     end
   end
+
+  # Raw requests suppress Req's body decoding, which would otherwise strip the
+  # structured error envelope off a failure response too.
+  defp decode_error_body(%Req.Response{body: body} = response) when is_binary(body) do
+    case Jason.decode(body) do
+      {:ok, decoded} -> %{response | body: decoded}
+      {:error, _reason} -> response
+    end
+  end
+
+  defp decode_error_body(response), do: response
 end

lib/archastro/query.ex

@@ -18,14 +18,21 @@ defmodule ArchAstro.SDK.Query do
     value
     |> ArchAstro.SDK.Codec.encode()
     |> Enum.flat_map(fn
-      {_key, :__archastro_unset__} -> []
-      {key, values} when is_list(values) -> Enum.map(values, &{key, scalar(&1)})
-      {key, item} -> [{key, scalar(item)}]
+      # A nil param means "not provided", never the string "null". The
+      # sentinel still arrives un-rewritten from Codec.encode's
+      # Map.from_struct fallback (structs without to_map/1).
+      {_key, value} when value in [nil, :__archastro_unset__] ->
+        []
+
+      {key, values} when is_list(values) ->
+        for value <- values, value not in [nil, :__archastro_unset__], do: {key, scalar(value)}
+
+      {key, item} ->
+        [{key, scalar(item)}]
     end)
     |> URI.encode_query()
   end
 
-  defp scalar(nil), do: "null"
   defp scalar(value) when is_binary(value), do: value
   defp scalar(value) when is_boolean(value) or is_number(value), do: to_string(value)
   defp scalar(value) when is_map(value) or is_list(value), do: Jason.encode!(value)

lib/archastro/socket.ex

@@ -12,6 +12,13 @@ defmodule ArchAstro.SDK.Socket do
 
   alias ArchAstro.SDK.{Channel, Client, Codec, Error, TokenServer}
 
+  # Pushes and joins the server never acknowledges (fire-and-forget events,
+  # callers that give up) would otherwise accumulate forever on long-lived
+  # sockets. Entries are swept once they are older than a full interval, so
+  # the effective grace period is one to two intervals — far above any sane
+  # Channel call timeout (default 5s).
+  @pending_sweep_interval_msec 60_000
+
   @type option ::
           {:name, GenServer.name()}
           | {:socket_path, String.t()}
@@ -29,6 +36,8 @@ defmodule ArchAstro.SDK.Socket do
   @impl Slipstream
   def init({client, opts}) do
     with {:ok, config} <- connection_config(client, opts) do
+      Process.send_after(self(), :archastro_sweep_pending, @pending_sweep_interval_msec)
+
       socket =
         new_socket()
         |> assign(
@@ -108,9 +117,15 @@ defmodule ArchAstro.SDK.Socket do
   end
 
   def handle_call({:archastro_push, topic, event, payload, descriptor}, from, socket) do
-    case push(socket, topic, event, Codec.encode(payload)) do
+    case push(socket, topic, event, payload) do
       {:ok, ref} ->
-        pending = %{from: from, topic: topic, descriptor: descriptor}
+        pending = %{
+          from: from,
+          topic: topic,
+          descriptor: descriptor,
+          inserted_at: System.monotonic_time(:millisecond)
+        }
+
         {:noreply, update(socket, :pending_pushes, &Map.put(&1, ref, pending))}
 
       {:error, reason} ->
@@ -175,9 +190,10 @@ defmodule ArchAstro.SDK.Socket do
   defp queue_join(socket, topic, payload, module, descriptor, from, nil) do
     pending = %{
       froms: [from],
-      payload: Codec.encode(payload),
+      payload: payload,
       module: module,
-      descriptor: descriptor
+      descriptor: descriptor,
+      inserted_at: System.monotonic_time(:millisecond)
     }
 
     update(socket, :pending_joins, &Map.put(&1, topic, pending))
@@ -208,22 +224,41 @@ defmodule ArchAstro.SDK.Socket do
   def handle_join(topic, response, socket) do
     case Map.pop(socket.assigns.pending_joins, topic) do
       {nil, pending_joins} ->
-        {:ok, assign(socket, :pending_joins, pending_joins)}
+        socket = assign(socket, :pending_joins, pending_joins)
+
+        # No waiter and no established channel means the join outlived
+        # everyone who wanted it (swept, or its callers gave up). Leaving
+        # returns the topic to a closed state; stranding it as joined-but-
+        # unowned would make every later join a silent no-op.
+        if Map.has_key?(socket.assigns.channels, topic) do
+          {:ok, socket}
+        else
+          {:ok, if(joined?(socket, topic), do: leave(socket, topic), else: socket)}
+        end
 
       {pending, pending_joins} ->
-        channel = %Channel{
-          socket: self(),
-          topic: topic,
-          module: pending.module,
-          join_response: Codec.decode(response, pending.descriptor)
-        }
-
-        Enum.each(pending.froms, &GenServer.reply(&1, {:ok, channel}))
-
-        {:ok,
-         socket
-         |> assign(:pending_joins, pending_joins)
-         |> update(:channels, &Map.put(&1, topic, channel))}
+        case safe_decode(response, pending.descriptor, %{topic: topic, kind: :join}) do
+          {:ok, join_response} ->
+            channel = %Channel{
+              socket: self(),
+              topic: topic,
+              module: pending.module,
+              join_response: join_response
+            }
+
+            Enum.each(pending.froms, &GenServer.reply(&1, {:ok, channel}))
+
+            {:ok,
+             socket
+             |> assign(:pending_joins, pending_joins)
+             |> update(:channels, &Map.put(&1, topic, channel))}
+
+          {:error, error} ->
+            Enum.each(pending.froms, &GenServer.reply(&1, {:error, error}))
+            socket = assign(socket, :pending_joins, pending_joins)
+            socket = if joined?(socket, topic), do: leave(socket, topic), else: socket
+            {:ok, socket}
+        end
     end
   end
 
@@ -234,7 +269,7 @@ defmodule ArchAstro.SDK.Socket do
         {:ok, assign(socket, :pending_pushes, pending)}
 
       {request, pending} ->
-        GenServer.reply(request.from, decode_reply(reply, request.descriptor))
+        GenServer.reply(request.from, decode_reply(reply, request))
         {:ok, assign(socket, :pending_pushes, pending)}
     end
   end
@@ -244,13 +279,55 @@ defmodule ArchAstro.SDK.Socket do
     channel = socket.assigns.channels[topic]
 
     Enum.each(socket.assigns.subscriptions[{topic, event}] || [], fn subscription ->
-      decoded = Codec.decode(message, subscription.descriptor)
-      send(subscription.subscriber, {:archastro_channel, channel, event, decoded})
+      case safe_decode(message, subscription.descriptor, %{
+             topic: topic,
+             event: event,
+             kind: :message
+           }) do
+        {:ok, decoded} ->
+          send(subscription.subscriber, {:archastro_channel, channel, event, decoded})
+
+        {:error, _error} ->
+          :ok
+      end
     end)
 
     {:ok, socket}
   end
 
+  @impl Slipstream
+  def handle_info(:archastro_sweep_pending, socket) do
+    Process.send_after(self(), :archastro_sweep_pending, @pending_sweep_interval_msec)
+    cutoff = System.monotonic_time(:millisecond) - @pending_sweep_interval_msec
+
+    {expired_pushes, live_pushes} =
+      Enum.split_with(socket.assigns.pending_pushes, fn {_ref, request} ->
+        request.inserted_at <= cutoff
+      end)
+
+    Enum.each(expired_pushes, fn {_ref, request} ->
+      GenServer.reply(request.from, {:error, no_reply_error("push")})
+    end)
+
+    {expired_joins, live_joins} =
+      Enum.split_with(socket.assigns.pending_joins, fn {_topic, pending} ->
+        pending.inserted_at <= cutoff
+      end)
+
+    Enum.each(expired_joins, fn {_topic, pending} ->
+      Enum.each(pending.froms, &GenServer.reply(&1, {:error, no_reply_error("join")}))
+    end)
+
+    {:noreply,
+     socket
+     |> assign(:pending_pushes, Map.new(live_pushes))
+     |> assign(:pending_joins, Map.new(live_joins))}
+  end
+
+  defp no_reply_error(kind) do
+    %Error{message: "channel #{kind} was never acknowledged", code: "no_reply"}
+  end
+
   @impl Slipstream
   def handle_topic_close(topic, reason, socket) do
     error = Error.channel(reason)
@@ -362,14 +439,35 @@ defmodule ArchAstro.SDK.Socket do
     end)
   end
 
-  defp decode_reply({:ok, value}, descriptor), do: {:ok, Codec.decode(value, descriptor)}
-  defp decode_reply({:error, reason}, _descriptor), do: {:error, Error.channel(reason)}
+  defp decode_reply({:ok, value}, request), do: safe_decode_reply(value, request)
+  defp decode_reply({:error, reason}, _request), do: {:error, Error.channel(reason)}
+
+  # Slipstream collapses a payload-less `{:reply, :error, socket}` to the bare
+  # atom :error; it is an error verdict, not a payload to decode.
+  defp decode_reply(:error, _request), do: {:error, Error.channel(%{})}
 
-  defp decode_reply(%{"status" => "ok", "response" => value}, descriptor),
-    do: {:ok, Codec.decode(value, descriptor)}
+  defp decode_reply(%{"status" => "ok", "response" => value}, request),
+    do: safe_decode_reply(value, request)
 
-  defp decode_reply(%{"status" => "error", "response" => reason}, _descriptor),
+  defp decode_reply(%{"status" => "error", "response" => reason}, _request),
     do: {:error, Error.channel(reason)}
 
-  defp decode_reply(value, descriptor), do: {:ok, Codec.decode(value, descriptor)}
+  defp decode_reply(value, request), do: safe_decode_reply(value, request)
+
+  defp safe_decode_reply(value, request),
+    do: safe_decode(value, request.descriptor, %{topic: request.topic, kind: :reply})
+
+  defp safe_decode(value, descriptor, metadata) do
+    {:ok, Codec.decode(value, descriptor)}
+  rescue
+    exception ->
+      :telemetry.execute([:archastro, :channel, :decode_failure], %{}, metadata)
+
+      {:error,
+       %Error{
+         message: "ArchAstro channel payload failed to decode",
+         code: "decode_failure",
+         reason: exception
+       }}
+  end
 end

lib/archastro/token_server/default.ex

@@ -89,7 +89,7 @@ defmodule ArchAstro.SDK.TokenServer.Default do
        table: table,
        base_url:
          String.trim_trailing(Keyword.get(opts, :base_url, "https://platform.archastro.ai"), "/"),
-       req: Keyword.get(opts, :req, Req.new(retry: false)),
+       req: Keyword.get(opts, :req, Req.new(retry: false, redirect: false)),
        store: opts[:store],
        inflight: %{}
      }}

test/codec_test.exs

@@ -82,4 +82,16 @@ defmodule ArchAstro.SDK.CodecTest do
     descriptor = {:union, [:datetime, {:enum, ["not-a-date"]}]}
     assert "not-a-date" = ArchAstro.SDK.Codec.decode("not-a-date", descriptor)
   end
+
+  test "encoding a NaiveDateTime raises a clear error instead of leaking a struct" do
+    assert_raise ArgumentError, ~r/NaiveDateTime.*DateTime/s, fn ->
+      ArchAstro.SDK.Codec.encode(%{"scheduled_at" => ~N[2026-08-16 12:00:00]})
+    end
+  end
+
+  test "encoding a Time raises a clear error instead of leaking a struct" do
+    assert_raise ArgumentError, ~r/Time/, fn ->
+      ArchAstro.SDK.Codec.encode(~T[12:00:00])
+    end
+  end
 end