Skip to content

Repository navigation

feat(realtime): add opt-in bounded transport recovery #1221

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
markstuart-oai merged 10 commits into from
Oct 18, 2026
Merged

feat(realtime): add opt-in bounded transport recovery #1221

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
10 commits
Select commit Hold shift + click to select a range
4b9778c
feat(realtime): add opt-in bounded transport recovery
markstuart-oai Oct 18, 2026
436916e
refactor(realtime): simplify recovery failure handling and docs
markstuart-oai Oct 18, 2026
29eb86d
style(realtime): apply repository Kotlin formatting
markstuart-oai Oct 18, 2026
2d0f603
fix(realtime): retain unsent commands across recovery
markstuart-oai Oct 18, 2026
7e59829
fix(realtime): isolate retries and inherit write timeouts
markstuart-oai Oct 18, 2026
7111be8
fix(realtime): resolve send admission before disconnect
markstuart-oai Oct 18, 2026
87bcafb
fix(realtime): preserve recovery across transport races
markstuart-oai Oct 18, 2026
0790f76
fix(realtime): classify pre-write lifecycle rejections
markstuart-oai Oct 18, 2026
ce59fed
fix(realtime): cancel recovery with client lifecycle
markstuart-oai Oct 18, 2026
5742687
test(realtime): reuse fake WebSocket transport setup
markstuart-oai Oct 18, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
21 changes: 21 additions & 0 deletions README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2149,6 +2149,27 @@ We take backwards-compatibility seriously and work hard to ensure you can rely o

We are keen for your feedback; please open an [issue](https://www.github.com/openai/openai-java/issues) with questions, bugs, or suggestions.

## Reconnecting to Realtime

Set `RealtimeWebSocketOptions.builder().reconnect(true).build()` to reconnect
after a temporary connection failure. This option is off by default. It does not
support transcription sessions or connections to an existing call. The SDK does
not retry normal closures, rejected credentials, or TLS failures.

Reconnecting keeps the same Java object but starts a new server session. The SDK
refreshes credentials for each attempt and allows five reconnect attempts in
total by default. Use `maxReconnectAttempts` to change this limit and
`reconnectDelayMillis` to set the initial delay. Delays grow to eight seconds.
Closing the connection cancels pending retries.

By default, the SDK rejects sends while reconnecting. Set `maxPendingBytes` to
hold unsent commands up to that size in bytes. Use `onReconnected` to restore any
session settings and conversation state your application needs. Pause other
application sends and wait for in-flight sends to finish before restoring state;
the callback does not block them. Then call `flushPending()` to send the waiting commands. To remove them without sending,
call `takePending()` instead. The SDK does not resend commands it already tried
to send. A successful send does not confirm that the server received the command.

## Contributing

Please share bug reports and feature requests through [GitHub issues](https://github.com/openai/openai-java/issues).
Expand Down
62 changes: 34 additions & 28 deletions openai-java-client-okhttp/src/main/kotlin/com/openai/client/okhttp/BoundedWebSocket.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,9 @@ package com.openai.client.okhttp
import com.openai.core.RequestOptions
import com.openai.core.http.WebSocketClient
import com.openai.core.http.WebSocketHandshakeException
import com.openai.core.http.WebSocketWriteNotAttempted
import java.io.IOException
import java.net.ProtocolException
import java.time.Duration
import java.util.concurrent.CompletableFuture
import java.util.concurrent.TimeUnit
Expand Down Expand Up @@ -108,7 +110,7 @@ internal object BoundedWebSocket {
) {
chain.call().cancel()
response.close()
throw IOException("WebSocket extensions are disabled")
throw ProtocolException("WebSocket extensions are disabled")
}
response
}
Expand Down Expand Up @@ -197,9 +199,10 @@ internal object BoundedWebSocket {
}
val deadline = System.nanoTime() + timeout.toNanos()
while (true) {
check(!ended.get() && closeDeadline.get() == null) {
"WebSocket is closed or closing"
}
if (ended.get() || closeDeadline.get() != null)
throw WebSocketWriteNotAttempted.Closed(
"WebSocket is closed or closing"
)
if (webSocket.queueSize() == 0L) return
val remaining = deadline - System.nanoTime()
if (remaining <= 0L)
Expand All @@ -214,32 +217,31 @@ internal object BoundedWebSocket {

override fun send(text: String) {
synchronized(this) {
check(!ended.get()) { "WebSocket is closed" }
require(text.isNotEmpty()) {
"WebSocket command must not be empty"
}
if (ended.get())
throw WebSocketWriteNotAttempted.Closed(
"WebSocket is closed"
)
if (text.isEmpty())
throw WebSocketWriteNotAttempted.Message(
"WebSocket command must not be empty"
)
val size = text.utf8Size()
if (size > maxMessageBytes)
throw com.openai.core.http
.WebSocketWriteNotAttempted
.Message(
"WebSocket command exceeds maxMessageBytes"
)
throw WebSocketWriteNotAttempted.Message(
"WebSocket command exceeds maxMessageBytes"
)
if (size > MAX_SEND_BYTES)
throw com.openai.core.http
.WebSocketWriteNotAttempted
.Message(
"WebSocket command exceeds OkHttp's 16 MiB send limit"
)
throw WebSocketWriteNotAttempted.Message(
"WebSocket command exceeds OkHttp's 16 MiB send limit"
)
if (webSocket.queueSize() != 0L)
throw com.openai.core.http
.WebSocketWriteNotAttempted
.Busy(
"WebSocket send buffer is full; command was not queued"
)
check(webSocket.send(text)) {
"WebSocket is closed; command was not queued"
}
throw WebSocketWriteNotAttempted.Busy(
"WebSocket send buffer is full; command was not queued"
)
if (!webSocket.send(text))
throw WebSocketWriteNotAttempted.Closed(
"WebSocket is closed; command was not queued"
)
}
}

Expand All @@ -258,13 +260,17 @@ internal object BoundedWebSocket {
synchronized(callbacks) {
if (ended.get()) return
if (text.utf8Size() > maxMessageBytes) {
fail(IOException("WebSocket message exceeds maxMessageBytes"))
fail(
ProtocolException(
"WebSocket message exceeds maxMessageBytes"
)
)
} else listener.onMessage(text)
}
}

override fun onMessage(webSocket: WebSocket, bytes: ByteString) {
fail(IOException("Unexpected binary WebSocket message"))
fail(ProtocolException("Unexpected binary WebSocket message"))
}

override fun onClosing(webSocket: WebSocket, code: Int, reason: String) {
Expand Down
Loading
Loading