From d86e7a84baeea4b1eadc874a322bb7c22d32128d Mon Sep 17 00:00:00 2001 From: UtkarshUsername Date: Fri, 16 Oct 2026 19:22:11 +0530 Subject: [PATCH 1/6] fix(client): deliver SSE events and close abandoned streams --- lib/opencode/client.ex | 23 ++++-- test/opencode/client_test.exs | 134 ++++++++++++++++++++++++++++++++++ 2 files changed, 151 insertions(+), 6 deletions(-) create mode 100644 test/opencode/client_test.exs diff --git a/lib/opencode/client.ex b/lib/opencode/client.ex index b0c9a04..8e69341 100644 --- a/lib/opencode/client.ex +++ b/lib/opencode/client.ex @@ -79,7 +79,8 @@ defmodule OpenCode.Client do Stream.resource( fn -> ref = make_ref() - pid = spawn_link(fn -> sse_run(req, self(), ref) end) + consumer = self() + pid = spawn_link(fn -> sse_run(req, consumer, ref) end) %{ref: ref, pid: pid, buf: ""} end, fn state -> @@ -97,8 +98,12 @@ defmodule OpenCode.Client do end, fn state -> case Process.alive?(state.pid) do - true -> Process.exit(state.pid, :normal) - false -> :ok + true -> + Process.unlink(state.pid) + Process.exit(state.pid, :shutdown) + + false -> + :ok end end ) @@ -121,8 +126,12 @@ defmodule OpenCode.Client do count = length(parts) case count do - 0 -> {[], ""} - 1 -> {[], data} + 0 -> + {[], ""} + + 1 -> + {[], data} + _ -> events = parts |> Enum.drop(-1) |> Enum.flat_map(&decode_sse/1) {events, List.last(parts)} @@ -139,7 +148,9 @@ defmodule OpenCode.Client do |> String.trim() case data do - "" -> [] + "" -> + [] + _ -> case Jason.decode(data) do {:ok, val} -> [val] diff --git a/test/opencode/client_test.exs b/test/opencode/client_test.exs new file mode 100644 index 0000000..3c93c9c --- /dev/null +++ b/test/opencode/client_test.exs @@ -0,0 +1,134 @@ +defmodule OpenCode.ClientTest do + use ExUnit.Case, async: true + + test "sends JSON bodies, query parameters, custom headers, and encoded directories" do + {base_url, request} = serve(201, ~s({"id":"session-test"})) + + client = + OpenCode.create_client( + base_url: base_url, + directory: "/tmp/日本語", + headers: %{"authorization" => "Bearer test"}, + timeout: 2_000 + ) + + assert {:ok, %{"id" => "session-test"}} = + OpenCode.Generated.Operations.session_create(%{title: "test"}, Keyword.put(client, :workspace, "a b")) + + raw = Task.await(request) + ["POST", target, "HTTP/1.1" <> _] = String.split(raw, " ", parts: 3) + uri = URI.parse(target) + assert uri.path == "/session" + assert URI.decode_query(uri.query) == %{"workspace" => "a b", "directory" => "/tmp/日本語"} + assert raw =~ "authorization: Bearer test" + assert raw =~ "x-opencode-directory: /tmp/%E6%97%A5%E6%9C%AC%E8%AA%9E" + assert raw =~ ~s({"title":"test"}) + end + + test "returns HTTP failures with status and decoded body" do + {base_url, request} = serve(404, ~s({"error":"missing"})) + + assert {:error, {404, %{"error" => "missing"}}} = + OpenCode.Generated.Session.session_get("missing", base_url: base_url) + + assert Task.await(request) =~ "GET /session/missing HTTP/1.1" + end + + test "streams SSE events across chunks and finishes when the response closes" do + body = "data: {\"type\":\"first\"}\n\ndata: {\"type\":\"second\"}\n\n" + {base_url, request} = serve(200, body, "text/event-stream") + + assert {:ok, %{stream: stream}} = OpenCode.Generated.Operations.event_subscribe(base_url: base_url) + task = Task.async(fn -> Enum.to_list(stream) end) + assert [%{"type" => "first"}, %{"type" => "second"}] = Task.await(task, 2_000) + assert Task.await(request) =~ "accept: text/event-stream" + end + + test "stopping an SSE stream closes the HTTP connection" do + {:ok, listener} = :gen_tcp.listen(0, [:binary, active: false]) + {:ok, {_, port}} = :inet.sockname(listener) + + server = + Task.async(fn -> + try do + {:ok, socket} = :gen_tcp.accept(listener, 2_000) + + try do + read_request(socket, "") + + :gen_tcp.send(socket, [ + "HTTP/1.1 200 OK\r\ncontent-type: text/event-stream\r\nconnection: close\r\n\r\n", + "data: {\"type\":\"connected\"}\n\n" + ]) + + :gen_tcp.recv(socket, 0, 2_000) + after + :gen_tcp.close(socket) + end + after + :gen_tcp.close(listener) + end + end) + + {:ok, %{stream: stream}} = + OpenCode.Generated.Operations.event_subscribe(base_url: "http://127.0.0.1:#{port}") + + consumer = Task.async(fn -> Enum.take(stream, 1) end) + assert [%{"type" => "connected"}] = Task.await(consumer, 2_000) + assert {:error, :closed} = Task.await(server, 2_000) + end + + defp serve(status, body, content_type \\ "application/json") do + {:ok, listener} = :gen_tcp.listen(0, [:binary, active: false, reuseaddr: true]) + {:ok, {_, port}} = :inet.sockname(listener) + + task = + Task.async(fn -> + try do + {:ok, socket} = :gen_tcp.accept(listener, 5_000) + + try do + request = read_request(socket, "") + + :ok = + :gen_tcp.send(socket, [ + "HTTP/1.1 #{status} Response\r\n", + "content-type: #{content_type}\r\n", + "content-length: #{byte_size(body)}\r\n", + "connection: close\r\n\r\n" + ]) + + for chunk <- :binary.bin_to_list(body), do: :gen_tcp.send(socket, <>) + request + after + :gen_tcp.close(socket) + end + after + :gen_tcp.close(listener) + end + end) + + {"http://127.0.0.1:#{port}", task} + end + + defp read_request(socket, buffer) do + case String.split(buffer, "\r\n\r\n", parts: 2) do + [headers, body] -> + length = + case Regex.run(~r/content-length: (\d+)/i, headers) do + [_, length] -> String.to_integer(length) + nil -> 0 + end + + if byte_size(body) >= length, do: buffer, else: receive_request(socket, buffer) + + _ -> + receive_request(socket, buffer) + end + end + + defp receive_request(socket, buffer) do + {:ok, chunk} = :gen_tcp.recv(socket, 0, 2_000) + read_request(socket, buffer <> chunk) + end +end From 54a9d6eea2827b73be7c1de91c7c38894c8369a1 Mon Sep 17 00:00:00 2001 From: UtkarshUsername Date: Fri, 16 Oct 2026 19:22:11 +0530 Subject: [PATCH 2/6] test(sdk): cover session lifecycle and server operations --- test/opencode/server_test.exs | 49 ++++++++++++++++++++++++++++++++++- 1 file changed, 48 insertions(+), 1 deletion(-) diff --git a/test/opencode/server_test.exs b/test/opencode/server_test.exs index 85c287c..06a0042 100644 --- a/test/opencode/server_test.exs +++ b/test/opencode/server_test.exs @@ -1,6 +1,8 @@ defmodule OpenCode.ServerTest do use ExUnit.Case, async: false + @moduletag skip: is_nil(System.find_executable("opencode")) + setup do case System.find_executable("opencode") do nil -> @@ -42,13 +44,15 @@ defmodule OpenCode.ServerTest do _ -> System.put_env(key, val) end end) + + File.rm_rf!(base) end) {:ok, env: env} end end - test "create_server starts and is healthy" do + test "managed server supports health, config, sessions, errors, and events" do port = free_port() {:ok, server} = @@ -59,6 +63,49 @@ defmodule OpenCode.ServerTest do client = OpenCode.create_client(base_url: server.url) assert {:ok, %{"healthy" => true}} = OpenCode.Generated.Operations.global_health(client) + assert {:ok, config} = OpenCode.Generated.Operations.config_get(client) + assert is_map(config) + assert {:ok, projects} = OpenCode.Generated.Operations.project_list(client) + assert is_list(projects) + + assert {:ok, session} = OpenCode.Generated.Operations.session_create(%{title: "SDK integration"}, client) + session_id = session["id"] + assert is_binary(session_id) + + try do + assert {:ok, %{"id" => ^session_id}} = OpenCode.Generated.Session.session_get(session_id, client) + + assert {:ok, sessions} = + OpenCode.Generated.Operations.session_list(Keyword.merge(client, search: "SDK integration", limit: 5)) + + assert Enum.any?(sessions, &(&1["id"] == session_id)) + + assert {:ok, %{"title" => "Updated SDK session"}} = + OpenCode.Generated.Operations.session_update(session_id, %{title: "Updated SDK session"}, client) + + assert {:ok, %{"info" => %{"role" => "user"}, "parts" => parts}} = + OpenCode.Generated.Operations.session_prompt( + session_id, + %{ + noReply: true, + model: %{providerID: "anthropic", modelID: "claude-sonnet-4-20250514"}, + parts: [%{type: "text", text: "SDK integration message"}] + }, + client + ) + + assert Enum.any?(parts, &(&1["type"] == "text" && &1["text"] == "SDK integration message")) + assert {:ok, messages} = OpenCode.Generated.Operations.session_messages(session_id, client) + assert Enum.any?(messages, &(&1["info"]["role"] == "user")) + + assert {:ok, %{stream: stream}} = OpenCode.Generated.Operations.event_subscribe(client) + task = Task.async(fn -> Enum.take(stream, 1) end) + assert [%{"type" => "server.connected"}] = Task.await(task, 5_000) + after + assert {:ok, true} = OpenCode.Generated.Operations.session_delete(session_id, client) + end + + assert {:error, {404, _}} = OpenCode.Generated.Session.session_get(session_id, client) after OpenCode.close(%{server: server}) end From a7cec7a569210cfcdacc997ab8f8e4e6a934e7e0 Mon Sep 17 00:00:00 2001 From: UtkarshUsername Date: Fri, 16 Oct 2026 19:22:12 +0530 Subject: [PATCH 3/6] fix(ci): sync compatible v1 spec and validate releases --- .github/workflows/ci.yml | 24 ++++++++++++ .github/workflows/sync-upstream.yml | 35 ++++++++++++++++-- README.md | 8 ++++ lib/mix/tasks/opencode.gen.client.ex | 12 ++++++ priv/opencode_openapi.json | 55 +++++++++++++++++++++++++++- test/opencode/generation_test.exs | 16 ++++++++ 6 files changed, 145 insertions(+), 5 deletions(-) create mode 100644 .github/workflows/ci.yml create mode 100644 test/opencode/generation_test.exs diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml new file mode 100644 index 0000000..f784993 --- /dev/null +++ b/.github/workflows/ci.yml @@ -0,0 +1,24 @@ +name: CI + +on: + push: + pull_request: + workflow_dispatch: + +permissions: + contents: read + +jobs: + test: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v6 + - uses: erlef/setup-beam@v1 + with: + elixir-version: "1.18" + otp-version: "27" + - run: mix deps.get + - run: npm install --global opencode-ai@1.19.3 + - run: mix compile --warnings-as-errors + - run: mix test --warnings-as-errors + - run: MIX_ENV=prod mix compile --warnings-as-errors diff --git a/.github/workflows/sync-upstream.yml b/.github/workflows/sync-upstream.yml index b6419e4..4c3095a 100644 --- a/.github/workflows/sync-upstream.yml +++ b/.github/workflows/sync-upstream.yml @@ -6,6 +6,10 @@ on: - cron: "0 6 * * *" workflow_dispatch: # Allow manual trigger +concurrency: + group: sync-upstream + cancel-in-progress: false + jobs: sync: runs-on: ubuntu-latest @@ -18,8 +22,20 @@ jobs: - name: Fetch upstream OpenAPI spec run: | - curl -sSL "https://raw.githubusercontent.com/anomalyco/opencode/dev/packages/sdk/openapi.json" \ - -o priv/opencode_openapi.json + curl --fail --silent --show-error --location --retry 3 --connect-timeout 10 --max-time 60 \ + "https://raw.githubusercontent.com/anomalyco/opencode/v1/packages/sdk/openapi.json" \ + -o "$RUNNER_TEMP/opencode_openapi.json" + python3 - "$RUNNER_TEMP/opencode_openapi.json" <<'PY' + import json + import sys + + with open(sys.argv[1]) as document: + spec = json.load(document) + assert spec.get("openapi", "").startswith("3."), "Expected an OpenAPI 3 document" + for path in ("/global/health", "/session", "/event"): + assert path in spec.get("paths", {}), f"Missing supported v1 route: {path}" + PY + mv "$RUNNER_TEMP/opencode_openapi.json" priv/opencode_openapi.json - name: Check for changes id: changes @@ -34,8 +50,8 @@ jobs: if: steps.changes.outputs.changed == 'true' uses: erlef/setup-beam@v1 with: - elixir-version: "1.15" - otp-version: "26" + elixir-version: "1.18" + otp-version: "27" - name: Install dependencies if: steps.changes.outputs.changed == 'true' @@ -45,6 +61,17 @@ jobs: if: steps.changes.outputs.changed == 'true' run: mix opencode.gen.client --spec priv/opencode_openapi.json + - name: Install OpenCode + if: steps.changes.outputs.changed == 'true' + run: npm install --global opencode-ai@1.19.3 + + - name: Validate SDK before release + if: steps.changes.outputs.changed == 'true' + run: | + mix compile --warnings-as-errors + mix test --warnings-as-errors + MIX_ENV=prod mix compile --warnings-as-errors + - name: Bump patch version if: steps.changes.outputs.changed == 'true' id: version diff --git a/README.md b/README.md index 407f59c..88942d4 100644 --- a/README.md +++ b/README.md @@ -378,12 +378,20 @@ API functions and types are documented in generated module docs, primarily: ## Regenerating The OpenAPI spec is at `priv/opencode_openapi.json`. +This SDK targets OpenCode's v1 API. The sync workflow follows +[`v1/packages/sdk/openapi.json`](https://github.com/anomalyco/opencode/blob/v1/packages/sdk/openapi.json). +Upstream `dev` now uses a different protocol API; its `packages/protocol/openapi.json` +is not a compatible replacement. Generation requires Elixir 1.18 or later. Regenerate the client with: ```bash mix opencode.gen.client --spec priv/opencode_openapi.json ``` +Run `mix test` to check the SDK. Server integration tests require the `opencode` +executable on `PATH`; CI uses OpenCode 1.19.3 and checks compilation and tests +before publishing an upstream sync. + ## Note This project is unofficial and is not affiliated with the OpenCode team. diff --git a/lib/mix/tasks/opencode.gen.client.ex b/lib/mix/tasks/opencode.gen.client.ex index 90dc43b..a98c9b6 100644 --- a/lib/mix/tasks/opencode.gen.client.ex +++ b/lib/mix/tasks/opencode.gen.client.ex @@ -51,6 +51,8 @@ defmodule Mix.Tasks.Opencode.Gen.Client do """) end + validate_spec!(spec_path) + Mix.shell().info("🔧 Generating OpenCode client from: #{spec_path}") Mix.shell().info("") @@ -76,6 +78,16 @@ defmodule Mix.Tasks.Opencode.Gen.Client do :ok end + defp validate_spec!(path) do + with {:ok, spec} <- Jason.decode(File.read!(path)), + %{"openapi" => "3." <> _, "paths" => paths} when is_map(paths) <- spec, + true <- map_size(paths) > 0 do + :ok + else + _ -> Mix.raise("Invalid OpenAPI spec at #{path}: expected an OpenAPI 3 JSON document with paths") + end + end + defp fix_missing_types do @generated_dir |> Path.join("*.ex") diff --git a/priv/opencode_openapi.json b/priv/opencode_openapi.json index e66d140..e064da6 100644 --- a/priv/opencode_openapi.json +++ b/priv/opencode_openapi.json @@ -20821,6 +20821,59 @@ "reasoning": { "type": "boolean" }, + "reasoning_options": { + "type": "array", + "items": { + "anyOf": [ + { + "type": "object", + "properties": { + "type": { + "type": "string", + "enum": ["effort"] + }, + "values": { + "type": "array", + "items": { + "type": "string" + } + } + }, + "required": ["type", "values"], + "additionalProperties": false + }, + { + "type": "object", + "properties": { + "type": { + "type": "string", + "enum": ["toggle"] + } + }, + "required": ["type"], + "additionalProperties": false + }, + { + "type": "object", + "properties": { + "type": { + "type": "string", + "enum": ["budget_tokens"] + }, + "min": { + "type": "number" + }, + "max": { + "type": "number" + } + }, + "required": ["type"], + "additionalProperties": false + } + ] + }, + "description": "Reasoning controls the model accepts, used to build its variants" + }, "temperature": { "type": "boolean" }, @@ -21263,7 +21316,7 @@ "enum": ["notify"] } ], - "description": "Automatically update to the latest version. Set to true to auto-update, false to disable, or 'notify' to show update notifications" + "description": "Check for new versions and show an update notice. Updates are never installed automatically; run /update or `opencode upgrade`. Set to false to disable update checks. true and 'notify' behave the same" }, "disabled_providers": { "type": "array", diff --git a/test/opencode/generation_test.exs b/test/opencode/generation_test.exs new file mode 100644 index 0000000..bdc0929 --- /dev/null +++ b/test/opencode/generation_test.exs @@ -0,0 +1,16 @@ +defmodule OpenCode.GenerationTest do + use ExUnit.Case, async: false + + test "rejects HTTP error pages and invalid specs before generation" do + path = Path.join(System.tmp_dir!(), "opencode-invalid-#{System.unique_integer([:positive])}.json") + on_exit(fn -> File.rm(path) end) + + for content <- ["404: Not Found", "{}", ~s({"openapi":"3.1.0","paths":{}})] do + File.write!(path, content) + + assert_raise Mix.Error, ~r/Invalid OpenAPI spec/, fn -> + Mix.Tasks.Opencode.Gen.Client.run(["--spec", path]) + end + end + end +end From fc03d23d0d4ba9622b22cde2235f1aa50be21228 Mon Sep 17 00:00:00 2001 From: UtkarshUsername Date: Fri, 16 Oct 2026 19:40:01 +0530 Subject: [PATCH 4/6] fix(client): terminate SSE failures and preserve HTTP errors --- lib/opencode/client.ex | 51 ++++++++++++++++++++++------------- test/opencode/client_test.exs | 41 ++++++++++++++++++++++++++-- 2 files changed, 72 insertions(+), 20 deletions(-) diff --git a/lib/opencode/client.ex b/lib/opencode/client.ex index 8e69341..af6f26b 100644 --- a/lib/opencode/client.ex +++ b/lib/opencode/client.ex @@ -81,20 +81,24 @@ defmodule OpenCode.Client do ref = make_ref() consumer = self() pid = spawn_link(fn -> sse_run(req, consumer, ref) end) - %{ref: ref, pid: pid, buf: ""} + %{ref: ref, pid: pid, buf: "", done?: false} end, - fn state -> - receive do - {ref, :data, data} when ref == state.ref -> - {events, buf} = parse_sse(state.buf <> data) - {events, %{state | buf: buf}} - - {ref, :done} when ref == state.ref -> - {:halt, state} - - {ref, :error, error} when ref == state.ref -> - {[{:error, error}], state} - end + fn + %{done?: true} = state -> + {:halt, state} + + state -> + receive do + {ref, :data, data} when ref == state.ref -> + {events, buf} = parse_sse(state.buf <> data) + {events, %{state | buf: buf}} + + {ref, :done} when ref == state.ref -> + {:halt, state} + + {ref, :error, error} when ref == state.ref -> + {[{:error, error}], %{state | done?: true}} + end end, fn state -> case Process.alive?(state.pid) do @@ -111,13 +115,24 @@ defmodule OpenCode.Client do defp sse_run(req, pid, ref) do case Req.request(req, - into: fn {:data, data}, acc -> - send(pid, {ref, :data, data}) - {:cont, acc} + retry: false, + into: fn {:data, data}, {request, response} -> + if response.status in 200..299 do + send(pid, {ref, :data, data}) + {:cont, {request, response}} + else + {:cont, {request, %{response | body: response.body <> data}}} + end end ) do - {:ok, _} -> send(pid, {ref, :done}) - {:error, error} -> send(pid, {ref, :error, error}) + {:ok, %{status: status}} when status in 200..299 -> + send(pid, {ref, :done}) + + {:ok, %{status: status, body: body}} -> + send(pid, {ref, :error, {status, body}}) + + {:error, error} -> + send(pid, {ref, :error, error}) end end diff --git a/test/opencode/client_test.exs b/test/opencode/client_test.exs index 3c93c9c..55733dd 100644 --- a/test/opencode/client_test.exs +++ b/test/opencode/client_test.exs @@ -78,7 +78,44 @@ defmodule OpenCode.ClientTest do assert {:error, :closed} = Task.await(server, 2_000) end - defp serve(status, body, content_type \\ "application/json") do + test "SSE HTTP errors preserve status and body without emitting events" do + for {status, body, content_type, expected} <- [ + {401, ~s({"error":"unauthorized"}), "application/json", %{"error" => "unauthorized"}}, + {500, "server failed", "text/plain", "server failed"}, + {403, "data: {\"type\":\"forbidden\"}\n\n", "text/event-stream", "data: {\"type\":\"forbidden\"}\n\n"} + ] do + {base_url, request} = serve(status, body, content_type) + {:ok, %{stream: stream}} = OpenCode.Generated.Operations.event_subscribe(base_url: base_url) + consumer = Task.async(fn -> Enum.to_list(stream) end) + + assert [{:error, {^status, ^expected}}] = Task.await(consumer, 2_000) + Task.await(request) + end + end + + test "SSE connection failures emit one transport error and terminate" do + {:ok, listener} = :gen_tcp.listen(0, [:binary, active: false]) + {:ok, {_, port}} = :inet.sockname(listener) + :gen_tcp.close(listener) + + {:ok, %{stream: stream}} = + OpenCode.Generated.Operations.event_subscribe(base_url: "http://127.0.0.1:#{port}") + + consumer = Task.async(fn -> Enum.to_list(stream) end) + assert [{:error, %Req.TransportError{reason: :econnrefused}}] = Task.await(consumer, 2_000) + end + + test "SSE mid-stream failures preserve delivered events and then terminate" do + body = "data: {\"type\":\"first\"}\n\n" + {base_url, request} = serve(200, body, "text/event-stream", byte_size(body) + 100) + {:ok, %{stream: stream}} = OpenCode.Generated.Operations.event_subscribe(base_url: base_url) + consumer = Task.async(fn -> Enum.to_list(stream) end) + + assert [%{"type" => "first"}, {:error, %Req.TransportError{}}] = Task.await(consumer, 2_000) + Task.await(request) + end + + defp serve(status, body, content_type \\ "application/json", content_length \\ nil) do {:ok, listener} = :gen_tcp.listen(0, [:binary, active: false, reuseaddr: true]) {:ok, {_, port}} = :inet.sockname(listener) @@ -94,7 +131,7 @@ defmodule OpenCode.ClientTest do :gen_tcp.send(socket, [ "HTTP/1.1 #{status} Response\r\n", "content-type: #{content_type}\r\n", - "content-length: #{byte_size(body)}\r\n", + "content-length: #{content_length || byte_size(body)}\r\n", "connection: close\r\n\r\n" ]) From 504c0e0fbd122d05521d5e26412c4e9c04fdb28c Mon Sep 17 00:00:00 2001 From: UtkarshUsername Date: Fri, 16 Oct 2026 19:54:52 +0530 Subject: [PATCH 5/6] fix(client): drain queued SSE messages when streams stop --- lib/opencode/client.ex | 32 ++++++++++++------ test/opencode/client_test.exs | 61 +++++++++++++++++++++++++++++++++++ 2 files changed, 83 insertions(+), 10 deletions(-) diff --git a/lib/opencode/client.ex b/lib/opencode/client.ex index af6f26b..d973bc3 100644 --- a/lib/opencode/client.ex +++ b/lib/opencode/client.ex @@ -100,19 +100,31 @@ defmodule OpenCode.Client do {[{:error, error}], %{state | done?: true}} end end, - fn state -> - case Process.alive?(state.pid) do - true -> - Process.unlink(state.pid) - Process.exit(state.pid, :shutdown) - - false -> - :ok - end - end + &sse_stop/1 ) end + defp sse_stop(%{pid: pid, ref: ref}) do + Process.unlink(pid) + monitor = Process.monitor(pid) + Process.exit(pid, :kill) + + receive do + {:DOWN, ^monitor, :process, ^pid, _reason} -> :ok + end + + drain_sse(ref) + end + + defp drain_sse(ref) do + receive do + {^ref, kind, _data} when kind in [:data, :error] -> drain_sse(ref) + {^ref, :done} -> drain_sse(ref) + after + 0 -> :ok + end + end + defp sse_run(req, pid, ref) do case Req.request(req, retry: false, diff --git a/test/opencode/client_test.exs b/test/opencode/client_test.exs index 55733dd..23d1ef7 100644 --- a/test/opencode/client_test.exs +++ b/test/opencode/client_test.exs @@ -78,6 +78,67 @@ defmodule OpenCode.ClientTest do assert {:error, :closed} = Task.await(server, 2_000) end + test "stopping an SSE stream drains queued messages and preserves unrelated messages" do + {:ok, listener} = :gen_tcp.listen(0, [:binary, active: false]) + {:ok, {_, port}} = :inet.sockname(listener) + first = "data: {\"type\":\"first\"}\n\n" + remaining = "data: {\"type\":\"second\"}\n\ndata: {\"type\":\"third\"}\n\n" + + server = + Task.async(fn -> + try do + {:ok, socket} = :gen_tcp.accept(listener, 2_000) + + try do + read_request(socket, "") + + :gen_tcp.send(socket, [ + "HTTP/1.1 200 OK\r\ncontent-type: text/event-stream\r\n", + "content-length: #{byte_size(first) + byte_size(remaining)}\r\nconnection: close\r\n\r\n", + first + ]) + + receive do + :finish -> :gen_tcp.send(socket, remaining) + after + 2_000 -> flunk("Consumer did not receive the first event") + end + after + :gen_tcp.close(socket) + end + after + :gen_tcp.close(listener) + end + end) + + {:ok, %{stream: stream}} = + OpenCode.Generated.Operations.event_subscribe(base_url: "http://127.0.0.1:#{port}") + + send(self(), :unrelated) + unrelated_ref = make_ref() + send(self(), {unrelated_ref, :data, "another stream"}) + + events = + stream + |> Stream.each(fn _event -> + send(server.pid, :finish) + + receive do + {reference, :done} = message when is_reference(reference) -> send(self(), message) + after + 2_000 -> flunk("SSE worker did not finish") + end + end) + |> Enum.take(1) + + assert events == [%{"type" => "first"}] + Task.await(server) + assert_received :unrelated + assert_received {^unrelated_ref, :data, "another stream"} + refute_received {_reference, :data, _} + refute_received {_reference, :done} + end + test "SSE HTTP errors preserve status and body without emitting events" do for {status, body, content_type, expected} <- [ {401, ~s({"error":"unauthorized"}), "application/json", %{"error" => "unauthorized"}}, From 3acca04575b63b2e6d4c085c473c1147dd243d20 Mon Sep 17 00:00:00 2001 From: UtkarshUsername Date: Fri, 16 Oct 2026 20:11:42 +0530 Subject: [PATCH 6/6] ci: test against the latest supported OpenCode release --- .github/workflows/ci.yml | 2 +- .github/workflows/sync-upstream.yml | 2 +- README.md | 2 +- 3 files changed, 3 insertions(+), 3 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index f784993..b7cb5fb 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -18,7 +18,7 @@ jobs: elixir-version: "1.18" otp-version: "27" - run: mix deps.get - - run: npm install --global opencode-ai@1.19.3 + - run: npm install --global opencode-ai@1 - run: mix compile --warnings-as-errors - run: mix test --warnings-as-errors - run: MIX_ENV=prod mix compile --warnings-as-errors diff --git a/.github/workflows/sync-upstream.yml b/.github/workflows/sync-upstream.yml index 4c3095a..0fdf40c 100644 --- a/.github/workflows/sync-upstream.yml +++ b/.github/workflows/sync-upstream.yml @@ -63,7 +63,7 @@ jobs: - name: Install OpenCode if: steps.changes.outputs.changed == 'true' - run: npm install --global opencode-ai@1.19.3 + run: npm install --global opencode-ai@1 - name: Validate SDK before release if: steps.changes.outputs.changed == 'true' diff --git a/README.md b/README.md index 88942d4..9cdf15f 100644 --- a/README.md +++ b/README.md @@ -389,7 +389,7 @@ mix opencode.gen.client --spec priv/opencode_openapi.json ``` Run `mix test` to check the SDK. Server integration tests require the `opencode` -executable on `PATH`; CI uses OpenCode 1.19.3 and checks compilation and tests +executable on `PATH`; CI uses the latest published OpenCode 1.x release and checks compilation and tests before publishing an upstream sync. ## Note