From bc425b50265cfc71228c29fe265dfae02f93e101 Mon Sep 17 00:00:00 2001 From: Rob Masson Date: Sun, 16 Aug 2026 10:50:34 -0700 Subject: [PATCH 01/12] fix(codec): load struct modules before checking for to_map/1 Codec.encode/1 gated the struct path on function_exported?(module, :to_map, 1), which never loads the module. When a payload struct's module was not yet loaded (normal for compile-time struct literals under interactive-mode code loading, e.g. mix test / iex), encode fell back to Map.from_struct and returned without recursing: the :__archastro_unset__ sentinel reached the wire as a literal string, renamed fields kept their struct keys (alias_ instead of "alias"), keys stayed atoms, and nested structs and DateTimes went un-encoded. The fallback re-fires independently at every nesting level, so the fix belongs in the encode clause itself, mirroring the existing Code.ensure_loaded?/1 guard on the decode-side {:ref, module} matcher. Regression test encodes every generated Input/Params struct with all-unset optionals under purged modules and asserts wire cleanliness, plus a renamed-field wire-name check. Co-Authored-By: Claude Fable 5 --- lib/archastro/codec.ex | 2 +- test/codec_unloaded_modules_test.exs | 79 ++++++++++++++++++++++++++++ 2 files changed, 80 insertions(+), 1 deletion(-) create mode 100644 test/codec_unloaded_modules_test.exs diff --git a/lib/archastro/codec.ex b/lib/archastro/codec.ex index 58d5016..239557e 100644 --- a/lib/archastro/codec.ex +++ b/lib/archastro/codec.ex @@ -74,7 +74,7 @@ defmodule ArchAstro.SDK.Codec do def encode(%Date{} = value), do: Date.to_iso8601(value) 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 diff --git a/test/codec_unloaded_modules_test.exs b/test/codec_unloaded_modules_test.exs new file mode 100644 index 0000000..532e99b --- /dev/null +++ b/test/codec_unloaded_modules_test.exs @@ -0,0 +1,79 @@ +# Copyright (c) 2026 ArchAstro Inc. Licensed under the MIT License. + +defmodule ArchAstro.SDK.CodecUnloadedModulesTest do + # Purging module code is VM-global state; keep this suite out of the + # concurrent test phase. + use ExUnit.Case, async: false + + alias ArchAstro.SDK.Codec + + @sentinel :__archastro_unset__ + + test "generated input/params structs encode identically when their modules are not loaded" do + modules = generated_input_modules() + assert length(modules) > 150 + + for module <- modules do + {:module, ^module} = Code.ensure_loaded(module) + value = struct(module) + expected = Codec.encode(value) + + unload(module) + actual = Codec.encode(value) + + assert actual == expected, + "#{inspect(module)} encoded differently when its module was not loaded: " <> + "#{inspect(actual)} vs #{inspect(expected)}" + + assert_wire_clean(actual, module) + {:module, ^module} = Code.ensure_loaded(module) + end + end + + test "renamed fields keep their wire names when the module is not loaded" do + module = ArchAstro.SDK.Types.Operations.PostApiV1AuthRegister.AuthInput + {:module, ^module} = Code.ensure_loaded(module) + value = struct(module, alias_: "my-alias") + + unload(module) + encoded = Codec.encode(value) + {:module, ^module} = Code.ensure_loaded(module) + + assert encoded["alias"] == "my-alias" + refute Map.has_key?(encoded, "alias_") + refute Map.has_key?(encoded, :alias_) + end + + defp generated_input_modules do + {:ok, modules} = :application.get_key(:archastro, :modules) + + Enum.filter(modules, fn module -> + name = Atom.to_string(module) + + String.starts_with?(name, "Elixir.ArchAstro.SDK.") and + Regex.match?(~r/\.(?:[A-Za-z0-9]*Input|Params)$/, name) + end) + end + + defp unload(module) do + :code.purge(module) + :code.delete(module) + refute function_exported?(module, :to_map, 1) + end + + defp assert_wire_clean(value, module) when is_map(value) do + Enum.each(value, fn {key, item} -> + assert is_binary(key), "#{inspect(module)} leaked non-string map key #{inspect(key)}" + assert_wire_clean(item, module) + end) + end + + defp assert_wire_clean(value, module) when is_list(value) do + Enum.each(value, &assert_wire_clean(&1, module)) + end + + defp assert_wire_clean(value, module) do + refute value in [@sentinel, "__archastro_unset__"], + "#{inspect(module)} leaked the unset sentinel onto the wire" + end +end From 013388900d839b8b6daf6ec47658249ab5e6eb3f Mon Sep 17 00:00:00 2001 From: Rob Masson Date: Sun, 16 Aug 2026 10:51:50 -0700 Subject: [PATCH 02/12] fix(http): stop Req from auto-decoding raw response bodies Raw operations (raw: true) return the Req.Response so callers get the body exactly as served, but Req's decode_body step ran first: a raw blob served with Content-Type: application/json arrived as a decoded map instead of the raw binary. Combined with the generator emitting decode: :string for GET /trajectories/{trajectory}/contents, that op raised on every successful non-empty fetch. Pass decode_body: false for raw requests so the body arrives untouched. Co-Authored-By: Claude Fable 5 --- lib/archastro/http.ex | 5 +++++ test/http_test.exs | 44 +++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 49 insertions(+) create mode 100644 test/http_test.exs diff --git a/lib/archastro/http.ex b/lib/archastro/http.ex index a9aaaf0..c2938f4 100644 --- a/lib/archastro/http.ex +++ b/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)) diff --git a/test/http_test.exs b/test/http_test.exs new file mode 100644 index 0000000..5bf513b --- /dev/null +++ b/test/http_test.exs @@ -0,0 +1,44 @@ +# Copyright (c) 2026 ArchAstro Inc. Licensed under the MIT License. + +defmodule ArchAstro.SDK.HTTPTest do + use ExUnit.Case, async: true + + alias ArchAstro.SDK.{Client, TokenServer} + + defp client(plug) do + server = + start_supervised!( + {TokenServer.Default, mode: {:system_user, publishable_key: "pk", access_token: "token"}} + ) + + {:ok, client} = Client.for_server(server, req: Req.new(plug: plug)) + client + end + + defp json_plug(body) do + fn conn -> + conn + |> Plug.Conn.put_resp_content_type("application/json") + |> Plug.Conn.send_resp(200, body) + end + end + + test "raw requests deliver an application/json body as the raw binary" do + raw_body = ~s({"steps":[{"n":1}],"note":"blob"}) + client = client(json_plug(raw_body)) + + assert {:ok, %Req.Response{} = response} = + ArchAstro.SDK.HTTP.request(client, :get, "/api/v1/trajectories/tra_1/contents", + raw: true + ) + + assert response.body == raw_body + end + + test "typed requests still auto-decode application/json bodies" do + client = client(json_plug(~s({"note":"blob"}))) + + assert {:ok, %{"note" => "blob"}} = + ArchAstro.SDK.HTTP.request(client, :get, "/api/v1/things/thing_1", decode: :unknown) + end +end From de39f9e74aeee0904bd661ab2f3de86459a84759 Mon Sep 17 00:00:00 2001 From: Rob Masson Date: Sun, 16 Aug 2026 10:55:00 -0700 Subject: [PATCH 03/12] harden(socket): survive undecodable channel payloads A naked Codec.decode in the Socket GenServer meant one malformed broadcast, reply, or join response raised inside the socket process, killing every channel, subscription, pending push, and the linked owner. A payload-less {:reply, :error, socket} from the server (collapsed to the bare atom :error by Slipstream) hit the same decode path and had the same blast radius. - broadcasts: drop the message for that subscriber and emit [:archastro, :channel, :decode_failure] telemetry - replies: resolve the push with {:error, %Error{code: "decode_failure"}} - join responses: fail the waiting joiners, leave the topic, keep the socket alive - bare :error replies: resolve as a channel error instead of decoding Co-Authored-By: Claude Fable 5 --- lib/archastro/socket.ex | 83 ++++++++++++++++++++-------- test/socket_test.exs | 116 ++++++++++++++++++++++++++++++++++++++++ 2 files changed, 177 insertions(+), 22 deletions(-) diff --git a/lib/archastro/socket.ex b/lib/archastro/socket.ex index cd2ad02..dc6f480 100644 --- a/lib/archastro/socket.ex +++ b/lib/archastro/socket.ex @@ -211,19 +211,28 @@ defmodule ArchAstro.SDK.Socket do {:ok, assign(socket, :pending_joins, pending_joins)} {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 +243,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,8 +253,17 @@ 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} @@ -362,14 +380,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)} - defp decode_reply(%{"status" => "ok", "response" => value}, descriptor), - do: {:ok, Codec.decode(value, descriptor)} + # 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" => "error", "response" => reason}, _descriptor), + defp decode_reply(%{"status" => "ok", "response" => value}, request), + do: safe_decode_reply(value, request) + + 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 diff --git a/test/socket_test.exs b/test/socket_test.exs index abf939f..20a0e27 100644 --- a/test/socket_test.exs +++ b/test/socket_test.exs @@ -212,4 +212,120 @@ defmodule ArchAstro.SDK.SocketTest do assert left_socket.assigns.pending_leaves == %{} refute Map.has_key?(left_socket.assigns.channels, "room:1") end + + describe "decode failures do not crash the socket" do + setup do + handler_id = "socket-decode-failure-#{inspect(make_ref())}" + + :telemetry.attach( + handler_id, + [:archastro, :channel, :decode_failure], + fn event, _measurements, metadata, pid -> send(pid, {:telemetry, event, metadata}) end, + self() + ) + + on_exit(fn -> :telemetry.detach(handler_id) end) + :ok + end + + test "a malformed broadcast is dropped with telemetry instead of raising" do + channel = %ArchAstro.SDK.Channel{socket: self(), topic: "room:1", module: __MODULE__} + + socket = + Slipstream.Socket.new() + |> Slipstream.Socket.assign(:channels, %{"room:1" => channel}) + |> Slipstream.Socket.assign(:subscriptions, %{ + {"room:1", "evt"} => [%{subscriber: self(), descriptor: :string}] + }) + + assert {:ok, _socket} = + ArchAstro.SDK.Socket.handle_message("room:1", "evt", %{"bad" => true}, socket) + + refute_received {:archastro_channel, _channel, _event, _payload} + + assert_received {:telemetry, [:archastro, :channel, :decode_failure], + %{topic: "room:1", event: "evt", kind: :message}} + end + + test "a well-formed broadcast is still delivered" do + channel = %ArchAstro.SDK.Channel{socket: self(), topic: "room:1", module: __MODULE__} + + socket = + Slipstream.Socket.new() + |> Slipstream.Socket.assign(:channels, %{"room:1" => channel}) + |> Slipstream.Socket.assign(:subscriptions, %{ + {"room:1", "evt"} => [%{subscriber: self(), descriptor: :string}] + }) + + assert {:ok, _socket} = + ArchAstro.SDK.Socket.handle_message("room:1", "evt", "hello", socket) + + assert_received {:archastro_channel, ^channel, "evt", "hello"} + end + + test "an undecodable reply fails only that push" do + tag = make_ref() + + socket = + Slipstream.Socket.new() + |> Slipstream.Socket.assign(:pending_pushes, %{ + "1" => %{from: {self(), tag}, topic: "room:1", descriptor: :string} + }) + + assert {:ok, updated} = + ArchAstro.SDK.Socket.handle_reply( + "1", + %{"status" => "ok", "response" => 42}, + socket + ) + + assert updated.assigns.pending_pushes == %{} + assert_receive {^tag, {:error, %ArchAstro.SDK.Error{code: "decode_failure"}}} + + assert_received {:telemetry, [:archastro, :channel, :decode_failure], + %{topic: "room:1", kind: :reply}} + end + + test "an empty error reply resolves the push as a channel error" do + tag = make_ref() + + socket = + Slipstream.Socket.new() + |> Slipstream.Socket.assign(:pending_pushes, %{ + "1" => %{from: {self(), tag}, topic: "room:1", descriptor: :string} + }) + + assert {:ok, updated} = ArchAstro.SDK.Socket.handle_reply("1", :error, socket) + + assert updated.assigns.pending_pushes == %{} + assert_receive {^tag, {:error, %ArchAstro.SDK.Error{} = error}} + assert error.message == "ArchAstro channel operation failed" + end + + test "an undecodable join response fails the join instead of the socket" do + tag = make_ref() + + socket = + Slipstream.Socket.new() + |> Slipstream.Socket.assign(:channels, %{}) + |> Slipstream.Socket.assign(:pending_joins, %{ + "room:1" => %{ + froms: [{self(), tag}], + payload: %{}, + module: __MODULE__, + descriptor: :string + } + }) + + assert {:ok, updated} = + ArchAstro.SDK.Socket.handle_join("room:1", %{"bad" => true}, socket) + + assert updated.assigns.pending_joins == %{} + refute Map.has_key?(updated.assigns.channels, "room:1") + assert_receive {^tag, {:error, %ArchAstro.SDK.Error{code: "decode_failure"}}} + + assert_received {:telemetry, [:archastro, :channel, :decode_failure], + %{topic: "room:1", kind: :join}} + end + end end From 5ff323a05ba70a4b72557f56e5a2747527ffbda7 Mon Sep 17 00:00:00 2001 From: Rob Masson Date: Sun, 16 Aug 2026 10:55:35 -0700 Subject: [PATCH 04/12] harden(codec): reject NaiveDateTime and Time inputs with a clear error A NaiveDateTime or Time in a request payload fell through encode's struct fallback into Map.from_struct, surfacing later as an opaque Jason tuple crash deep inside Req. Raise an ArgumentError at the encode boundary that names the value and the conversion the caller needs. Co-Authored-By: Claude Fable 5 --- lib/archastro/codec.ex | 13 +++++++++++++ test/codec_test.exs | 12 ++++++++++++ 2 files changed, 25 insertions(+) diff --git a/lib/archastro/codec.ex b/lib/archastro/codec.ex index 239557e..445ae9e 100644 --- a/lib/archastro/codec.ex +++ b/lib/archastro/codec.ex @@ -73,6 +73,19 @@ 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 Code.ensure_loaded?(module) and function_exported?(module, :to_map, 1), do: module.to_map(value), diff --git a/test/codec_test.exs b/test/codec_test.exs index 0aac0b3..e0e9446 100644 --- a/test/codec_test.exs +++ b/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 From 229ed371f35b8cf4ec1899caed6ea3027d4876fe Mon Sep 17 00:00:00 2001 From: Rob Masson Date: Sun, 16 Aug 2026 10:56:26 -0700 Subject: [PATCH 05/12] harden(client): disable redirect-following on default requests The internal default Req.new(retry: false) in Client.build and TokenServer.Default left Req's redirect-following on. A 302 from the platform silently rewrote a token-refresh POST into a body-less GET against the Location target. A 3xx is never a valid platform response; surface it as the error it is. Co-Authored-By: Claude Fable 5 --- lib/archastro/client.ex | 2 +- lib/archastro/token_server/default.ex | 2 +- test/token_server_test.exs | 11 +++++++++++ 3 files changed, 13 insertions(+), 2 deletions(-) diff --git a/lib/archastro/client.ex b/lib/archastro/client.ex index f7f601a..969efc6 100644 --- a/lib/archastro/client.ex +++ b/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, %{}) }} diff --git a/lib/archastro/token_server/default.ex b/lib/archastro/token_server/default.ex index 59f034d..a10346e 100644 --- a/lib/archastro/token_server/default.ex +++ b/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: %{} }} diff --git a/test/token_server_test.exs b/test/token_server_test.exs index b153a68..9bd6ecb 100644 --- a/test/token_server_test.exs +++ b/test/token_server_test.exs @@ -537,4 +537,15 @@ defmodule ArchAstro.SDK.TokenServerTest do assert {:ok, authorization} = TokenServer.authorization(client.token_binding) assert {"x-archastro-api-key", "sk_test"} in authorization.headers end + + test "default requests never follow redirects" do + # A 3xx from the platform is always wrong; following it can silently turn a + # token-refresh POST into a body-less GET against the Location target. + server = start_supervised!({TokenServer.Default, mode: {:secret_key, "sk_test"}}) + + assert :sys.get_state(server).req.options[:redirect] == false + + assert {:ok, client} = Client.for_server(server) + assert client.request.options[:redirect] == false + end end From 92a5698c94b3f6a43b328d8f997de8bfc3d1654b Mon Sep 17 00:00:00 2001 From: Rob Masson Date: Sun, 16 Aug 2026 10:57:19 -0700 Subject: [PATCH 06/12] harden(query): drop nil params instead of sending the string null MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A nil scalar in a plain-map query serialized as the literal string "null" (scalar/1 had a nil clause), and the unset-sentinel filter was dead code because Codec.encode rewrites sentinels to nil before the filter runs. Treat nil — top-level or inside a repeated param — as "not provided" and omit it. Co-Authored-By: Claude Fable 5 --- lib/archastro/query.ex | 14 ++++++++++---- test/query_test.exs | 11 +++++++++++ 2 files changed, 21 insertions(+), 4 deletions(-) diff --git a/lib/archastro/query.ex b/lib/archastro/query.ex index cf542d3..06179cc 100644 --- a/lib/archastro/query.ex +++ b/lib/archastro/query.ex @@ -18,14 +18,20 @@ 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)}] + # Codec.encode has already rewritten unset sentinels to nil; a nil + # param means "not provided", never the string "null". + {_key, nil} -> + [] + + {key, values} when is_list(values) -> + for value <- values, value != nil, 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) diff --git a/test/query_test.exs b/test/query_test.exs index 8e444d3..e7f4f9d 100644 --- a/test/query_test.exs +++ b/test/query_test.exs @@ -32,4 +32,15 @@ defmodule ArchAstro.SDK.QueryTest do assert ArchAstro.SDK.Query.append("https://example.test/path", %Params{tags: []}) == "https://example.test/path" end + + test "omits nil params instead of sending the string null" do + assert ArchAstro.SDK.Query.encode(%{"cursor" => nil, "limit" => 5}) == "limit=5" + + assert ArchAstro.SDK.Query.append("https://example.test/path", %{"cursor" => nil}) == + "https://example.test/path" + end + + test "omits nil elements inside repeated params" do + assert ArchAstro.SDK.Query.encode(%{"tags" => ["one", nil, "two"]}) == "tags=one&tags=two" + end end From d97a3622659a78bec0e31043fc83fd137fc012f4 Mon Sep 17 00:00:00 2001 From: Rob Masson Date: Sun, 16 Aug 2026 10:59:15 -0700 Subject: [PATCH 07/12] harden(socket): sweep pushes and joins that are never acknowledged MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit pending_pushes and pending_joins entries had no expiry: a push for a fire-and-forget event (or a join the server never acks) parked its entry forever on a long-lived socket. Entries now carry an insertion timestamp and a periodic sweep fails anything older than a full interval (60s, so an effective 60-120s grace — far above the 5s default Channel call timeout) with {:error, %Error{code: "no_reply"}}. Replies to callers that already timed out are no-ops. Co-Authored-By: Claude Fable 5 --- lib/archastro/socket.ex | 53 +++++++++++++++++++++++++++++++-- test/socket_test.exs | 66 +++++++++++++++++++++++++++++++++++++++++ 2 files changed, 117 insertions(+), 2 deletions(-) diff --git a/lib/archastro/socket.ex b/lib/archastro/socket.ex index dc6f480..7e49e28 100644 --- a/lib/archastro/socket.ex +++ b/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( @@ -110,7 +119,13 @@ defmodule ArchAstro.SDK.Socket do def handle_call({:archastro_push, topic, event, payload, descriptor}, from, socket) do case push(socket, topic, event, Codec.encode(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} -> @@ -177,7 +192,8 @@ defmodule ArchAstro.SDK.Socket do froms: [from], payload: Codec.encode(payload), module: module, - descriptor: descriptor + descriptor: descriptor, + inserted_at: System.monotonic_time(:millisecond) } update(socket, :pending_joins, &Map.put(&1, topic, pending)) @@ -269,6 +285,39 @@ defmodule ArchAstro.SDK.Socket do {: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) diff --git a/test/socket_test.exs b/test/socket_test.exs index 20a0e27..f1fd40c 100644 --- a/test/socket_test.exs +++ b/test/socket_test.exs @@ -213,6 +213,72 @@ defmodule ArchAstro.SDK.SocketTest do refute Map.has_key?(left_socket.assigns.channels, "room:1") end + describe "pending entry sweep" do + test "sweeps pushes and joins that were never acknowledged, keeping fresh ones" do + stale_push_tag = make_ref() + fresh_push_tag = make_ref() + stale_join_tag = make_ref() + + now = System.monotonic_time(:millisecond) + stale = now - 200_000 + + socket = + Slipstream.Socket.new() + |> Slipstream.Socket.assign(:channels, %{}) + |> Slipstream.Socket.assign(:subscriptions, %{}) + |> Slipstream.Socket.assign(:pending_pushes, %{ + "1" => %{ + from: {self(), stale_push_tag}, + topic: "room:1", + descriptor: :string, + inserted_at: stale + }, + "2" => %{ + from: {self(), fresh_push_tag}, + topic: "room:1", + descriptor: :string, + inserted_at: now + } + }) + |> Slipstream.Socket.assign(:pending_joins, %{ + "room:2" => %{ + froms: [{self(), stale_join_tag}], + payload: %{}, + module: __MODULE__, + descriptor: :string, + inserted_at: stale + } + }) + + assert {:noreply, swept} = + ArchAstro.SDK.Socket.handle_info(:archastro_sweep_pending, socket) + + assert Map.keys(swept.assigns.pending_pushes) == ["2"] + assert swept.assigns.pending_joins == %{} + + assert_receive {^stale_push_tag, {:error, %ArchAstro.SDK.Error{code: "no_reply"}}} + assert_receive {^stale_join_tag, {:error, %ArchAstro.SDK.Error{code: "no_reply"}}} + refute_received {^fresh_push_tag, _reply} + end + + test "new pushes and queued joins carry an insertion timestamp" do + socket = + Slipstream.Socket.new() + |> Slipstream.Socket.assign(:channels, %{}) + |> Slipstream.Socket.assign(:pending_joins, %{}) + |> Slipstream.Socket.assign(:pending_leaves, %{"room:1" => true}) + + assert {:noreply, queued} = + ArchAstro.SDK.Socket.handle_call( + {:archastro_join, "room:1", %{}, __MODULE__, {:object, []}}, + {self(), make_ref()}, + socket + ) + + assert is_integer(queued.assigns.pending_joins["room:1"].inserted_at) + end + end + describe "decode failures do not crash the socket" do setup do handler_id = "socket-decode-failure-#{inspect(make_ref())}" From 68497fa6d79519aa06d8885e7ac80dd2bbae9174 Mon Sep 17 00:00:00 2001 From: Rob Masson Date: Sun, 16 Aug 2026 11:10:29 -0700 Subject: [PATCH 08/12] ci: build the contract harness from archastro-openapi main on PRs Pull-request runs pinned the harness checkout to a long-dead SDK development branch that predates the harness's nullable-schema normalization, so it exits at boot against the 0.3.3 spec and all 15 channel-contract tests fail on any PR (main pushes already use main and are green). Verified locally: harness built from main passes all 15 against this branch. Co-Authored-By: Claude Fable 5 --- .github/workflows/ci.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 8afd9cd..44d253a 100644 --- a/.github/workflows/ci.yml +++ b/.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) From ee482c4ae7ed6f74e9f4b9c68844de37ab662acf Mon Sep 17 00:00:00 2001 From: Rob Masson Date: Sun, 16 Aug 2026 11:14:51 -0700 Subject: [PATCH 09/12] fix(channel): encode outbound payloads in the caller, not the socket MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Codec.encode ran inside the Socket GenServer for pushes and joins, so any unencodable payload raised there and killed the shared socket with every other channel, subscription, and in-flight push on it. That was survivable while encode was total, but this branch makes it raise on NaiveDateTime/Time — an easy value to hand it, since Ecto timestamps() defaults to :naive_datetime. Encode in Channel.push/join instead, so a bad payload fails only its own caller. Found by an adversarial audit subagent over this branch. Co-Authored-By: Claude Fable 5 --- lib/archastro/channel.ex | 25 ++++++++++++++++--------- lib/archastro/socket.ex | 4 ++-- test/socket_test.exs | 29 +++++++++++++++++++++++++++++ 3 files changed, 47 insertions(+), 11 deletions(-) diff --git a/lib/archastro/channel.ex b/lib/archastro/channel.ex index d0404b9..c202fe7 100644 --- a/lib/archastro/channel.ex +++ b/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) diff --git a/lib/archastro/socket.ex b/lib/archastro/socket.ex index 7e49e28..9a98ded 100644 --- a/lib/archastro/socket.ex +++ b/lib/archastro/socket.ex @@ -117,7 +117,7 @@ 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, @@ -190,7 +190,7 @@ 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, inserted_at: System.monotonic_time(:millisecond) diff --git a/test/socket_test.exs b/test/socket_test.exs index f1fd40c..84175e5 100644 --- a/test/socket_test.exs +++ b/test/socket_test.exs @@ -213,6 +213,35 @@ defmodule ArchAstro.SDK.SocketTest do refute Map.has_key?(left_socket.assigns.channels, "room:1") end + describe "outbound payload encoding" do + test "encode failures raise in the caller process, never inside the socket" do + test_pid = self() + # A stand-in socket that reports anything it receives and never replies, + # so a payload reaching the socket process is visible as a message. + socket = spawn_link(fn -> relay_forever(test_pid) end) + channel = %ArchAstro.SDK.Channel{socket: socket, topic: "room:1", module: __MODULE__} + bad_payload = %{"scheduled_at" => ~N[2026-08-16 12:00:00]} + + assert_raise ArgumentError, ~r/NaiveDateTime/, fn -> + ArchAstro.SDK.Channel.push(channel, "evt", bad_payload, :string, 200) + end + + assert_raise ArgumentError, ~r/NaiveDateTime/, fn -> + ArchAstro.SDK.Channel.join(socket, "room:1", bad_payload, __MODULE__, :string, 200) + end + + refute_receive {:socket_received, _message}, 50 + end + end + + defp relay_forever(test_pid) do + receive do + message -> + send(test_pid, {:socket_received, message}) + relay_forever(test_pid) + end + end + describe "pending entry sweep" do test "sweeps pushes and joins that were never acknowledged, keeping fresh ones" do stale_push_tag = make_ref() From 2edc8dfc90ddbf44f8d8c3af3db0afe1e1db1f52 Mon Sep 17 00:00:00 2001 From: Rob Masson Date: Sun, 16 Aug 2026 11:15:49 -0700 Subject: [PATCH 10/12] fix(http): keep structured API errors on raw requests Suppressing Req's body decoding for raw requests also stripped the error envelope off failure responses: Error.from_response discards a non-map body, so a 404 carrying {"error": {"message", "code"}} degraded to the generic "ArchAstro API returned HTTP 404" with no code or details. Decode a binary body before building the error, the same way the SSE path already does; a non-JSON body is left alone. Found by an adversarial audit subagent over this branch. Co-Authored-By: Claude Fable 5 --- lib/archastro/http.ex | 13 ++++++++++++- test/http_test.exs | 32 ++++++++++++++++++++++++++++++++ 2 files changed, 44 insertions(+), 1 deletion(-) diff --git a/lib/archastro/http.ex b/lib/archastro/http.ex index c2938f4..ff30ca7 100644 --- a/lib/archastro/http.ex +++ b/lib/archastro/http.ex @@ -75,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], @@ -97,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 diff --git a/test/http_test.exs b/test/http_test.exs index 5bf513b..aa794b6 100644 --- a/test/http_test.exs +++ b/test/http_test.exs @@ -35,6 +35,38 @@ defmodule ArchAstro.SDK.HTTPTest do assert response.body == raw_body end + test "raw requests keep structured API errors on failure responses" do + error_plug = fn conn -> + conn + |> Plug.Conn.put_resp_content_type("application/json") + |> Plug.Conn.send_resp( + 404, + ~s({"error":{"message":"Trajectory not found","code":"not_found"}}) + ) + end + + client = client(error_plug) + + assert {:error, %ArchAstro.SDK.Error{} = error} = + ArchAstro.SDK.HTTP.request(client, :get, "/api/v1/trajectories/tra_1/contents", + raw: true + ) + + assert error.status == 404 + assert error.message == "Trajectory not found" + assert error.code == "not_found" + end + + test "raw requests tolerate a non-JSON failure body" do + error_plug = fn conn -> Plug.Conn.send_resp(conn, 502, "upstream exploded") end + client = client(error_plug) + + assert {:error, %ArchAstro.SDK.Error{status: 502} = error} = + ArchAstro.SDK.HTTP.request(client, :get, "/api/v1/files/file_1/avatar", raw: true) + + assert error.message == "ArchAstro API returned HTTP 502" + end + test "typed requests still auto-decode application/json bodies" do client = client(json_plug(~s({"note":"blob"}))) From bd6a693b21d6d3663f17cdcfa8cb200ed6c65e25 Mon Sep 17 00:00:00 2001 From: Rob Masson Date: Sun, 16 Aug 2026 11:16:15 -0700 Subject: [PATCH 11/12] fix(query): keep dropping unset sentinels, not just nils The previous commit removed the sentinel clause on the premise that Codec.encode always rewrites :__archastro_unset__ to nil first. That holds only for structs exporting to_map/1; the Map.from_struct fallback leaves sentinels intact, so a caller-built params struct following the SDK's own sentinel convention crashed in scalar/1 where it used to produce the correct URL. Drop both nil and the sentinel, top-level and inside repeated params. Found by an adversarial audit subagent over this branch. Co-Authored-By: Claude Fable 5 --- lib/archastro/query.ex | 9 +++++---- test/query_test.exs | 10 ++++++++++ 2 files changed, 15 insertions(+), 4 deletions(-) diff --git a/lib/archastro/query.ex b/lib/archastro/query.ex index 06179cc..0b8b046 100644 --- a/lib/archastro/query.ex +++ b/lib/archastro/query.ex @@ -18,13 +18,14 @@ defmodule ArchAstro.SDK.Query do value |> ArchAstro.SDK.Codec.encode() |> Enum.flat_map(fn - # Codec.encode has already rewritten unset sentinels to nil; a nil - # param means "not provided", never the string "null". - {_key, nil} -> + # 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 != nil, do: {key, scalar(value)} + for value <- values, value not in [nil, :__archastro_unset__], do: {key, scalar(value)} {key, item} -> [{key, scalar(item)}] diff --git a/test/query_test.exs b/test/query_test.exs index e7f4f9d..78ea8f2 100644 --- a/test/query_test.exs +++ b/test/query_test.exs @@ -43,4 +43,14 @@ defmodule ArchAstro.SDK.QueryTest do test "omits nil elements inside repeated params" do assert ArchAstro.SDK.Query.encode(%{"tags" => ["one", nil, "two"]}) == "tags=one&tags=two" end + + defmodule SentinelParams do + # A params struct with no to_map/1: Codec.encode falls back to + # Map.from_struct, which does not rewrite unset sentinels. + defstruct limit: :__archastro_unset__, cursor: :__archastro_unset__ + end + + test "omits unset sentinels that reach encoding un-rewritten" do + assert ArchAstro.SDK.Query.encode(%SentinelParams{limit: 5}) == "limit=5" + end end From 3c7702f19bee09b7effa072b9500b70a1ae16b4e Mon Sep 17 00:00:00 2001 From: Rob Masson Date: Sun, 16 Aug 2026 11:17:30 -0700 Subject: [PATCH 12/12] fix(socket): leave a topic whose join response has no waiter MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A join acked after its waiters are gone (swept, or callers that gave up) left Slipstream's join status at :joined while no channel existed in assigns. Slipstream's join/3 no-ops unless the status is nil or :closed, so every later join of that topic queued a pending entry that could never be satisfied — the topic stayed unjoinable until the socket reconnected. Leave the topic instead, returning it to a closed state. Established topics rejoined after a reconnect keep their channel and are not left. Found by an adversarial audit subagent over this branch. Co-Authored-By: Claude Fable 5 --- lib/archastro/socket.ex | 12 ++++++++++- test/socket_test.exs | 45 +++++++++++++++++++++++++++++++++++++++++ 2 files changed, 56 insertions(+), 1 deletion(-) diff --git a/lib/archastro/socket.ex b/lib/archastro/socket.ex index 9a98ded..1c4dea0 100644 --- a/lib/archastro/socket.ex +++ b/lib/archastro/socket.ex @@ -224,7 +224,17 @@ 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} -> case safe_decode(response, pending.descriptor, %{topic: topic, kind: :join}) do diff --git a/test/socket_test.exs b/test/socket_test.exs index 84175e5..a9c9ee0 100644 --- a/test/socket_test.exs +++ b/test/socket_test.exs @@ -308,6 +308,51 @@ defmodule ArchAstro.SDK.SocketTest do end end + describe "join responses with no waiter" do + test "a join acked after its waiters are gone leaves the topic instead of stranding it" do + socket = + Slipstream.Socket.new() + |> Map.put(:channel_pid, self()) + |> Map.put(:joins, %{ + "room:1" => %Slipstream.Socket.Join{ + topic: "room:1", + params: %{}, + status: :joined, + rejoin_counter: 0 + } + }) + |> Slipstream.Socket.assign(:channels, %{}) + |> Slipstream.Socket.assign(:pending_joins, %{}) + + assert {:ok, _socket} = ArchAstro.SDK.Socket.handle_join("room:1", %{}, socket) + + assert_receive {:__slipstream_command__, %Slipstream.Commands.LeaveTopic{topic: "room:1"}} + end + + test "an established topic rejoined after reconnect is kept, not left" do + channel = %ArchAstro.SDK.Channel{socket: self(), topic: "room:1", module: __MODULE__} + + socket = + Slipstream.Socket.new() + |> Map.put(:channel_pid, self()) + |> Map.put(:joins, %{ + "room:1" => %Slipstream.Socket.Join{ + topic: "room:1", + params: %{}, + status: :joined, + rejoin_counter: 1 + } + }) + |> Slipstream.Socket.assign(:channels, %{"room:1" => channel}) + |> Slipstream.Socket.assign(:pending_joins, %{}) + + assert {:ok, kept} = ArchAstro.SDK.Socket.handle_join("room:1", %{}, socket) + + assert Map.has_key?(kept.assigns.channels, "room:1") + refute_received {:__slipstream_command__, %Slipstream.Commands.LeaveTopic{}} + end + end + describe "decode failures do not crash the socket" do setup do handler_id = "socket-decode-failure-#{inspect(make_ref())}"