diff --git a/.env.live.example b/.env.live.example index 4717d54..bb15197 100644 --- a/.env.live.example +++ b/.env.live.example @@ -30,3 +30,20 @@ QODER_SA_MODEL=auto # Optional single-region factory input (QODER_SA_KEY), used only explicitly. QODER_SA_KEY= QODER_ACCESS_TOKEN= + +# Explicit Realtime acceptance covers both regions × PAT/SA. Use existing +# pre-enabled Templates; the suite never changes their Realtime settings. +QODER_FORWARD_CN_PAT= +QODER_FORWARD_GLOBAL_PAT= +# Forward-issued bound SATs for Realtime. Platform Credential SATs are currently +# rejected by Realtime; omitting these retains the platform SA Key check above. +QODER_REALTIME_CN_SAT= +QODER_REALTIME_GLOBAL_SAT= +QODER_REALTIME_CN_PAT_TEMPLATE_ID= +QODER_REALTIME_GLOBAL_PAT_TEMPLATE_ID= +QODER_REALTIME_CN_SA_TEMPLATE_ID= +QODER_REALTIME_GLOBAL_SA_TEMPLATE_ID= +QODER_REALTIME_LIVE_ALLOW_WRITE=false +QODER_REALTIME_LIVE_ALLOW_EXECUTION=false +# Optional raw speech PCM16 LE, 16 kHz mono, at most 30s including trailing silence. +QODER_REALTIME_LIVE_PCM_FILE= diff --git a/CHANGELOG.md b/CHANGELOG.md index 6161608..0487745 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,9 @@ existing `0.1.0` release; earlier development prereleases are not listed. - Forward admin and Identity SAT issuance and Forward SAT revocation, using explicit per-call SA Key authentication, typed issuance/revocation metadata, and no automatic retries. Existing platform credentials and PAT/SAT business authentication remain unchanged. +- Forward Realtime Beta: create Conversations with a stable idempotency key, connect over WebSocket using PAT/SAT or shared Credentials, exchange text and PCM16 audio, observe typed events and Work progress, report playback, interrupt, and confirm graceful closure. +- Realtime Conversation history with independent `before` pagination, persisted history IDs and preserved unknown event fields. Connections have separate stage timeouts, bounded receive queues, and explicit protocol failures; they do not follow redirects, reconnect, or replay input automatically. + ## [0.4.0] - 2026-10-12 ### Added diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 02cd61e..ef582f7 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -62,6 +62,40 @@ When adding or changing an endpoint: The fixtures are maintained manually. A passing fixture test proves consistency with this repository, not automatically with service routes or the other SDKs. +Realtime protocol regressions use a local WebSocket server, never an account or +model. Run `go test -race ./forward -run Realtime -count=1` when changing the +connection, shared upgrade authentication, history or event models. The +`forward/testdata/realtime-events.json` fixture records the `voice.realtime.v1` +contract and source snapshot. Real CN/Global Realtime checks additionally need a +pre-enabled Template and an allowed Identity; the standalone example can perform +create/ready/ping/close/history, with optional text or raw PCM input. It retains +Conversation IDs because no public deletion API exists. Mock tests do not prove +that either production region has deployed the protocol. + +`make test-live-realtime` separately verifies CN/Global × PAT/SA with the two +`QODER_REALTIME_LIVE_ALLOW_WRITE=true` and +`QODER_REALTIME_LIVE_ALLOW_EXECUTION=true` gates. Configure +`QODER_FORWARD_{CN,GLOBAL}_PAT`, `QODER_REALTIME_{CN,GLOBAL}_SAT` and the four +`QODER_REALTIME_{CN,GLOBAL}_{PAT,SA}_TEMPLATE_ID` values in `.env.live` or the +process environment. Realtime currently requires Forward-issued bound SATs; +the platform SAT exchanged by `ServiceAccountCredential` does not create the +Forward binding. Without an explicit regional SAT, the suite falls back to the +corresponding `QODER_SA_{CN,GLOBAL}_KEY` to retain this platform compatibility +check, which currently receives 403 from Realtime. Per region, one SA credential +source must exist. All required inputs are checked before any request. +The suite creates unique test Identities, creates/connects Conversations, +checks ready and pong, sends a short text without fixed model wording, decodes +output PCM, confirms graceful closure, and exercises filtered history at page +size 1. It finishes the associated Forward Sessions and deletes only its test +Identities; it retains/logs Conversation IDs and never edits the supplied +Templates. This entry consumes model usage. Optionally set +`QODER_REALTIME_LIVE_PCM_FILE` to a raw 16 kHz mono PCM16 little-endian speech +fixture (at most 30 seconds, including trailing silence). The file is checked +before requests; the additional audio round requires a user transcript and an +assistant transcript/audio response. Without this setting, the suite tests text +input and does not prove microphone ASR. The standalone example also supports +`-pcm`. + ## Compatibility conventions The SDK intentionally keeps Qoder-branded `X-Qoder-*` metadata headers and resumable session-event streams. Preserve those extensions unless the change explicitly revises the public contract. Breaking public API changes require a minor-version release while the SDK remains pre-1.0 and must include migration notes. diff --git a/Makefile b/Makefile index c45bbb3..920c7b2 100644 --- a/Makefile +++ b/Makefile @@ -74,7 +74,7 @@ test: test-unit test-contract test-unit: go test ./... go test -tags live -run 'Cleanup.*Offline$$' ./forward ./managed - go test -tags live -run '^TestSAConfiguration$$' ./integration + go test -tags live -run '^Test(SAConfiguration|RealtimeConfiguration)$$' ./integration test-contract: go test -v \ @@ -131,5 +131,9 @@ test-live-sa: test-live-sa-execution: LIVE_ENV_FILE="$(abspath $(LIVE_ENV_FILE))" go test -tags live -count=1 -v -run TestServiceAccountExecutionLive -timeout=15m ./integration +.PHONY: test-live-realtime +test-live-realtime: + LIVE_ENV_FILE="$(abspath $(LIVE_ENV_FILE))" go test -tags live -count=1 -v -run '^TestRealtimeLive$$' -timeout=25m ./integration + test-live-sa-tokens: LIVE_ENV_FILE="$(abspath $(LIVE_ENV_FILE))" go test -tags live -count=1 -v -run '^TestForwardServiceAccountTokensLive$$' -timeout=10m ./integration diff --git a/README.md b/README.md index 9c0ebb4..844a954 100644 --- a/README.md +++ b/README.md @@ -8,7 +8,7 @@ The SDK ships two clients that share authentication, transport, retries, error h | Mode | Package | Resources | | --- | --- | --- | -| Forward | `forward` | Identity, Template, Session, Schedule, Batch, Channel, Environment, File, Skill, Vault, MemoryStore, Model, Usage | +| Forward | `forward` | Identity, Template, Session, Schedule, Batch, Channel, Environment, File, Skill, Vault, MemoryStore, Model, Usage, Realtime | | Managed | `managed` | Agent, Session, Deployment, Dream, Environment, File, Skill, Vault, MemoryStore, Model | - [Installation](#installation) @@ -19,6 +19,7 @@ The SDK ships two clients that share authentication, transport, retries, error h - [Conversations](#conversations) - [System prompts and tools](#system-prompts-and-tools) - [Streaming](#streaming) +- [Realtime Beta](#realtime-beta) - [Request fields](#request-fields) - [Response objects](#response-objects) - [Error handling](#error-handling) @@ -855,6 +856,135 @@ func readUnknownField(session *forward.Session) string { ## Examples +### Realtime Beta + +`client.Realtime` connects to the public Forward gateway. Realtime must already +be enabled on the Template; this SDK does not change its Realtime configuration. +Creating a Conversation requires a stable `Idempotency-Key` (1–256 UTF-8 bytes) +for that operation. It stays in a header during retries, rather than the JSON body. + +```go +conversation, err := client.Realtime.Conversations.New(ctx, + forward.RealtimeConversationNewParams{ + IdentityID: identityID, + TemplateID: templateID, + IdempotencyKey: forward.String(operationKey), + }) +if err != nil { return err } +connection, err := client.Realtime.Connect(ctx, conversation.ID, + forward.RealtimeConnectParams{}) +if err != nil { return err } +defer connection.Close() +// Connect returns after voice.ready; that event and earlier voice.state events +// remain in the receive queue. ctx spans the connection lifetime. +if err := connection.SendText(ctx, "Hello"); err != nil { return err } +event, err := connection.Recv(ctx) +if err != nil { return err } +var transcript forward.RealtimeTranscriptPayload +if event.Type == "transcript.final" { + if err := event.DecodePayload(&transcript); err != nil { return err } +} +// Stop capture/playback before graceful close. Ordinary Close only releases +// transport resources; it does not confirm that history was saved. +closed, err := connection.CloseGracefully(ctx) +if err != nil { return err } +fmt.Println(closed.Outcome) +``` + +REST and each new WebSocket connection use the same PAT, direct SAT, or shared +`Credential`. A credential provider is called on each new connection; an +established socket cannot update its Authorization header. The current Realtime +service requires an SA Token bound through Forward's Token issuance APIs. +The platform SAT obtained by `ServiceAccountCredential` alone does not create +this binding and is currently rejected by Realtime. Use +`option.WithAuthMode("sa")` and `option.WithAuthToken(boundSAT)` for a directly +supplied bound SAT, or a custom Credential that supplies one. SDK transport +support does not grant additional Token permissions. Credentials stay in headers, +and upgrade redirects are rejected. Native `*http.Client` transports (including +TLS and proxy configuration), headers and middleware apply to the upgrade; +arbitrary custom HTTP doers are unsupported. The client's HTTP `Timeout` and +`option.WithRequestTimeout` do not limit a Realtime connection's lifetime. + +`RealtimeConnectParams` defaults to 30s handshake, 60s ready, 30s send, 10s +graceful close, 256 queued events / 4 MiB, and 1 MiB per received message. +Exceeding receive limits terminates the connection with an explicit error. +`Recv(ctx)` preserves order, including buffered events before a terminal error. +Known payload types cover transcripts, audio, state, errors, close and Work +events; `Payload`, `DecodePayload` and `RawJSON()` preserve unknown extensions. +The timestamp stays text, including nanosecond precision. `voice.replaced` +ends the old connection; a nonterminal `error` or `work.failed` remains an event. + +`AppendAudio` takes raw **16 kHz mono PCM16 little-endian** bytes, not WAV, and +sends Base64 in JSON text frames. Output is **24 kHz mono PCM16**; +`RealtimeEvent.Audio()` decodes `audio.delta`. The final client JSON message +must fit within 256 KiB; text is limited to 16 KiB after trimming whitespace. +Applications own sampling, playback and its lifecycle: `audio.done` means +generation finished, while `PlaybackStarted`, `PlaybackEnded` and +`PlaybackCancelled` report what the speaker actually did. Copy `work_id` and +`announcement_id` from the corresponding audio event into the receipt. +`Interrupt` does not stop an application's speaker by itself. + +The service limits a connection to 60 minutes. The SDK does not automatically +reconnect or replay text/audio; applications may reconnect to the same +Conversation. Closing a connection neither cancels background Work nor deletes +the Conversation. Graceful closure succeeds only after a matching request ID +and successful persistence outcome; timeouts and persistence failures stay errors. + +History has its own `conversation / events / page` envelope and `before` cursor, +and iterates towards older records. Historical `id` values are independent of +the WebSocket `event_id` and sequence. Filters are `message`, `work` or +`message,work`, and page size is 1–100. `HasNextPage` / `GetNextPage(ctx)` expose +manual paging; `HistoryAutoPaging` also crosses empty pages and detects cursor cycles. + +```go +history := client.Realtime.Conversations.HistoryAutoPaging(ctx, conversation.ID, + forward.RealtimeConversationHistoryParams{ + Types: forward.String("message,work"), Limit: forward.Int(50), + }) +for history.Next() { + event := history.Current() + fmt.Println(event.ID, event.Type, event.OccurredAt) +} +if err := history.Err(); err != nil { return err } +``` + +The standalone [Realtime example](examples/forward/realtime/main.go) uses +process environment variables and a pre-enabled Template/Identity. It defaults +to ping/close/history without sending model input. Optional `-text` and `-pcm` +use model quota; choose either text or raw speech PCM (at most 30 seconds, +including trailing silence), sent in 20 ms chunks. The demo does not play audio and +reports received output as cancelled. New Conversations remain stored because +the API exposes no Conversation deletion endpoint. + +```sh +# PAT defaults; SA uses the same QODER_AUTH_MODE/SA_KEY/ACCESS_TOKEN convention. +QODER_SA_REGION=cn go run ./examples/forward/realtime \ + -identity "$QODER_IDENTITY_ID" -template "$QODER_TEMPLATE_ID" \ + -idempotency-key "$QODER_IDEMPOTENCY_KEY" +# Supply a Forward-issued bound SAT through QODER_ACCESS_TOKEN; do not set QODER_SA_KEY. +QODER_AUTH_MODE=sa QODER_SA_REGION=global go run ./examples/forward/realtime \ + -conversation "$QODER_CONVERSATION_ID" -text "Hello" +go run ./examples/forward/realtime -conversation "$QODER_CONVERSATION_ID" -history-only +``` + +Maintainers can run `make test-live-realtime` for an explicit CN/Global × PAT/SA +acceptance suite. Configure the four pre-enabled +`QODER_REALTIME_{CN,GLOBAL}_{PAT,SA}_TEMPLATE_ID` values and matching +`QODER_FORWARD_{CN,GLOBAL}_PAT` / `QODER_REALTIME_{CN,GLOBAL}_SAT` credentials, then +set both `QODER_REALTIME_LIVE_ALLOW_WRITE=true` and +`QODER_REALTIME_LIVE_ALLOW_EXECUTION=true`. The suite validates all inputs +before creating test Identities, verifies text/output PCM and persisted history, +and cleans up its Sessions/Identities. It retains Conversation IDs and leaves +the supplied Templates intact. The SA cases use the supplied Forward-issued +bound SATs. Without them, they use `QODER_SA_{CN,GLOBAL}_KEY` through +`ServiceAccountCredential` to retain platform Token compatibility checks; the +current service rejects such unbound platform SATs. No Token issuance is +performed implicitly. Optionally set `QODER_REALTIME_LIVE_PCM_FILE` to +a raw 16 kHz mono PCM16 little-endian speech fixture of at most 30 seconds, +including trailing silence. It adds an audio round requiring a user transcript +and an assistant transcript/audio response. Without it, acceptance covers text +input; the standalone example also supports `-pcm`. + `examples/forward/` and `examples/managed/` hold 18 business programs, one `main.go` per scenario, each runnable directly. They read `.env.live` from the working directory — run them from the repository root, or point `-env` elsewhere. Copy [.env.live.example](.env.live.example) and configure PAT or SA authentication. Examples default to CN; `-region cn|global` overrides the configured Qoder hostname. ```sh diff --git a/convention/websocket.go b/convention/websocket.go new file mode 100644 index 0000000..cf5bd12 --- /dev/null +++ b/convention/websocket.go @@ -0,0 +1,46 @@ +package convention + +import ( + "fmt" + "net/http" +) + +// WebSocketHandshake prepares a single upgrade using the same headers, +// authentication and middleware as HTTP requests. RequestTimeout and the HTTP +// client's Timeout do not limit the lifetime of the upgraded connection. +// Custom HTTP doers cannot guarantee WebSocket upgrades; use a native http.Client. +func (cfg *RequestConfig) WebSocketHandshake() (*http.Client, *http.Request, error) { + if cfg.CustomHTTPDoer != nil { + return nil, nil, fmt.Errorf("qca: Realtime requires an *http.Client, not a custom HTTP doer") + } + base := cfg.BaseURL + if base == nil { + base = cfg.DefaultBaseURL + } + if base == nil || base.Host == "" || base.User != nil || base.RawQuery != "" || base.Fragment != "" || (base.Scheme != "http" && base.Scheme != "https") { + return nil, nil, fmt.Errorf("qca: Realtime base URL must be an HTTP(S) URL without user info, query or fragment") + } + request := cfg.Request.Clone(cfg.Request.Context()) + var err error + request.URL, err = base.Parse(request.URL.String()) + if err != nil { + return nil, nil, err + } + client := *cfg.HTTPClient + client.Timeout = 0 + client.CheckRedirect = func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse } + transport := client.Transport + if transport == nil { + transport = http.DefaultTransport + } + handler := middlewareNext(transport.RoundTrip) + for i := len(cfg.Middlewares) - 1; i >= 0; i-- { + handler = applyMiddleware(cfg.Middlewares[i], handler) + } + client.Transport = websocketRoundTrip(handler) + return &client, request, nil +} + +type websocketRoundTrip middlewareNext + +func (f websocketRoundTrip) RoundTrip(r *http.Request) (*http.Response, error) { return f(r) } diff --git a/docs/api/reference.md b/docs/api/reference.md index 62330c1..67c7651 100644 --- a/docs/api/reference.md +++ b/docs/api/reference.md @@ -410,6 +410,83 @@ import "github.com/QoderAI/qoder-cloud-agents-sdk-go/forward" - [type PermissionPolicyParam](<#PermissionPolicyParam>) - [func \(r PermissionPolicyParam\) MarshalJSON\(\) \(\[\]byte, error\)](<#PermissionPolicyParam.MarshalJSON>) - [func \(r \*PermissionPolicyParam\) UnmarshalJSON\(data \[\]byte\) error](<#PermissionPolicyParam.UnmarshalJSON>) +- [type RealtimeAudioConfig](<#RealtimeAudioConfig>) +- [type RealtimeAudioConfigParam](<#RealtimeAudioConfigParam>) + - [func \(r RealtimeAudioConfigParam\) MarshalJSON\(\) \(\[\]byte, error\)](<#RealtimeAudioConfigParam.MarshalJSON>) +- [type RealtimeAudioFormat](<#RealtimeAudioFormat>) +- [type RealtimeAudioOutputConfig](<#RealtimeAudioOutputConfig>) +- [type RealtimeAudioOutputConfigParam](<#RealtimeAudioOutputConfigParam>) + - [func \(r RealtimeAudioOutputConfigParam\) MarshalJSON\(\) \(\[\]byte, error\)](<#RealtimeAudioOutputConfigParam.MarshalJSON>) +- [type RealtimeAudioPayload](<#RealtimeAudioPayload>) +- [type RealtimeCapabilities](<#RealtimeCapabilities>) +- [type RealtimeClientEvent](<#RealtimeClientEvent>) +- [type RealtimeClosedPayload](<#RealtimeClosedPayload>) +- [type RealtimeConfig](<#RealtimeConfig>) +- [type RealtimeConfigParam](<#RealtimeConfigParam>) + - [func \(r RealtimeConfigParam\) MarshalJSON\(\) \(\[\]byte, error\)](<#RealtimeConfigParam.MarshalJSON>) +- [type RealtimeConnectParams](<#RealtimeConnectParams>) +- [type RealtimeConnection](<#RealtimeConnection>) + - [func \(c \*RealtimeConnection\) AppendAudio\(ctx context.Context, pcm \[\]byte\) error](<#RealtimeConnection.AppendAudio>) + - [func \(c \*RealtimeConnection\) Close\(\) error](<#RealtimeConnection.Close>) + - [func \(c \*RealtimeConnection\) CloseGracefully\(ctx context.Context\) \(\*RealtimeClosedPayload, error\)](<#RealtimeConnection.CloseGracefully>) + - [func \(c \*RealtimeConnection\) Err\(\) error](<#RealtimeConnection.Err>) + - [func \(c \*RealtimeConnection\) Interrupt\(ctx context.Context\) error](<#RealtimeConnection.Interrupt>) + - [func \(c \*RealtimeConnection\) Ping\(ctx context.Context\) error](<#RealtimeConnection.Ping>) + - [func \(c \*RealtimeConnection\) PlaybackCancelled\(ctx context.Context, r RealtimePlaybackReceipt\) error](<#RealtimeConnection.PlaybackCancelled>) + - [func \(c \*RealtimeConnection\) PlaybackEnded\(ctx context.Context, r RealtimePlaybackReceipt\) error](<#RealtimeConnection.PlaybackEnded>) + - [func \(c \*RealtimeConnection\) PlaybackStarted\(ctx context.Context, r RealtimePlaybackReceipt\) error](<#RealtimeConnection.PlaybackStarted>) + - [func \(c \*RealtimeConnection\) Recv\(ctx context.Context\) \(RealtimeEvent, error\)](<#RealtimeConnection.Recv>) + - [func \(c \*RealtimeConnection\) SendEvent\(ctx context.Context, event RealtimeClientEvent\) error](<#RealtimeConnection.SendEvent>) + - [func \(c \*RealtimeConnection\) SendText\(ctx context.Context, text string\) error](<#RealtimeConnection.SendText>) +- [type RealtimeConversation](<#RealtimeConversation>) +- [type RealtimeConversationHistoryParams](<#RealtimeConversationHistoryParams>) + - [func \(r RealtimeConversationHistoryParams\) URLQuery\(\) \(url.Values, error\)](<#RealtimeConversationHistoryParams.URLQuery>) +- [type RealtimeConversationNewParams](<#RealtimeConversationNewParams>) + - [func \(r RealtimeConversationNewParams\) MarshalJSON\(\) \(\[\]byte, error\)](<#RealtimeConversationNewParams.MarshalJSON>) +- [type RealtimeConversationService](<#RealtimeConversationService>) + - [func NewRealtimeConversationService\(opts ...option.RequestOption\) RealtimeConversationService](<#NewRealtimeConversationService>) + - [func \(r \*RealtimeConversationService\) History\(ctx context.Context, conversationID string, params RealtimeConversationHistoryParams, opts ...option.RequestOption\) \(res \*RealtimeHistoryPage, err error\)](<#RealtimeConversationService.History>) + - [func \(r \*RealtimeConversationService\) HistoryAutoPaging\(ctx context.Context, conversationID string, params RealtimeConversationHistoryParams, opts ...option.RequestOption\) \*RealtimeHistoryAutoPager](<#RealtimeConversationService.HistoryAutoPaging>) + - [func \(r \*RealtimeConversationService\) New\(ctx context.Context, params RealtimeConversationNewParams, opts ...option.RequestOption\) \(res \*RealtimeConversation, err error\)](<#RealtimeConversationService.New>) +- [type RealtimeErrorPayload](<#RealtimeErrorPayload>) +- [type RealtimeEvent](<#RealtimeEvent>) + - [func \(e RealtimeEvent\) Audio\(\) \(\[\]byte, error\)](<#RealtimeEvent.Audio>) + - [func \(e RealtimeEvent\) DecodePayload\(dst any\) error](<#RealtimeEvent.DecodePayload>) + - [func \(e RealtimeEvent\) RawJSON\(\) string](<#RealtimeEvent.RawJSON>) +- [type RealtimeHistoryAutoPager](<#RealtimeHistoryAutoPager>) + - [func \(p \*RealtimeHistoryAutoPager\) Current\(\) RealtimeHistoryEvent](<#RealtimeHistoryAutoPager.Current>) + - [func \(p \*RealtimeHistoryAutoPager\) Err\(\) error](<#RealtimeHistoryAutoPager.Err>) + - [func \(p \*RealtimeHistoryAutoPager\) Next\(\) bool](<#RealtimeHistoryAutoPager.Next>) +- [type RealtimeHistoryConversation](<#RealtimeHistoryConversation>) +- [type RealtimeHistoryEvent](<#RealtimeHistoryEvent>) + - [func \(e RealtimeHistoryEvent\) RawJSON\(\) string](<#RealtimeHistoryEvent.RawJSON>) + - [func \(e \*RealtimeHistoryEvent\) UnmarshalJSON\(b \[\]byte\) error](<#RealtimeHistoryEvent.UnmarshalJSON>) +- [type RealtimeHistoryPage](<#RealtimeHistoryPage>) + - [func \(p \*RealtimeHistoryPage\) GetNextPage\(ctx context.Context\) \(\*RealtimeHistoryPage, error\)](<#RealtimeHistoryPage.GetNextPage>) + - [func \(p \*RealtimeHistoryPage\) HasNextPage\(\) bool](<#RealtimeHistoryPage.HasNextPage>) +- [type RealtimeHistoryPageInfo](<#RealtimeHistoryPageInfo>) +- [type RealtimePlaybackReceipt](<#RealtimePlaybackReceipt>) +- [type RealtimeProtocolError](<#RealtimeProtocolError>) + - [func \(e \*RealtimeProtocolError\) Error\(\) string](<#RealtimeProtocolError.Error>) +- [type RealtimeReadyPayload](<#RealtimeReadyPayload>) +- [type RealtimeReasonPayload](<#RealtimeReasonPayload>) +- [type RealtimeServerError](<#RealtimeServerError>) + - [func \(e \*RealtimeServerError\) Error\(\) string](<#RealtimeServerError.Error>) +- [type RealtimeService](<#RealtimeService>) + - [func NewRealtimeService\(opts ...option.RequestOption\) RealtimeService](<#NewRealtimeService>) + - [func \(r \*RealtimeService\) Connect\(ctx context.Context, conversationID string, params RealtimeConnectParams, opts ...option.RequestOption\) \(\*RealtimeConnection, error\)](<#RealtimeService.Connect>) +- [type RealtimeStatePayload](<#RealtimeStatePayload>) +- [type RealtimeTranscriptPayload](<#RealtimeTranscriptPayload>) +- [type RealtimeTransportError](<#RealtimeTransportError>) + - [func \(e \*RealtimeTransportError\) Error\(\) string](<#RealtimeTransportError.Error>) + - [func \(e \*RealtimeTransportError\) Unwrap\(\) error](<#RealtimeTransportError.Unwrap>) +- [type RealtimeWorkAcceptedPayload](<#RealtimeWorkAcceptedPayload>) +- [type RealtimeWorkCompletedPayload](<#RealtimeWorkCompletedPayload>) +- [type RealtimeWorkError](<#RealtimeWorkError>) +- [type RealtimeWorkFailedPayload](<#RealtimeWorkFailedPayload>) +- [type RealtimeWorkMilestonePayload](<#RealtimeWorkMilestonePayload>) +- [type RealtimeWorkProgressPayload](<#RealtimeWorkProgressPayload>) +- [type RealtimeWorkRunningPayload](<#RealtimeWorkRunningPayload>) - [type ResourceBinding](<#ResourceBinding>) - [func \(r ResourceBinding\) RawJSON\(\) string](<#ResourceBinding.RawJSON>) - [func \(r \*ResourceBinding\) UnmarshalJSON\(data \[\]byte\) error](<#ResourceBinding.UnmarshalJSON>) @@ -782,6 +859,18 @@ import "github.com/QoderAI/qoder-cloud-agents-sdk-go/forward" const DefaultBaseURL = "https://api.qoder.com/api/v1/forward/" ``` + + +```go +const RealtimeMaxClientMessageBytes = 256 * 1024 +``` + + + +```go +const RealtimeProtocolVersion = "voice.realtime.v1" +``` + ## func [Bool]() @@ -2063,6 +2152,7 @@ type Client struct { Usage UsageService Models ModelService ServiceAccountTokens ServiceAccountTokenService + Realtime RealtimeService } ``` @@ -5844,6 +5934,868 @@ func (r *PermissionPolicyParam) UnmarshalJSON(data []byte) error + +## type [RealtimeAudioConfig]() + + + +```go +type RealtimeAudioConfig struct { + Output RealtimeAudioOutputConfig `json:"output"` +} +``` + + +## type [RealtimeAudioConfigParam]() + + + +```go +type RealtimeAudioConfigParam struct { + Output RealtimeAudioOutputConfigParam `json:"output,omitzero"` + // contains filtered or unexported fields +} +``` + + +### func \(RealtimeAudioConfigParam\) [MarshalJSON]() + +```go +func (r RealtimeAudioConfigParam) MarshalJSON() ([]byte, error) +``` + + + + +## type [RealtimeAudioFormat]() + + + +```go +type RealtimeAudioFormat struct { + Format string `json:"format"` + SampleRate int `json:"sample_rate"` + Channels int `json:"channels"` +} +``` + + +## type [RealtimeAudioOutputConfig]() + + + +```go +type RealtimeAudioOutputConfig struct { + Voice string `json:"voice"` +} +``` + + +## type [RealtimeAudioOutputConfigParam]() + + + +```go +type RealtimeAudioOutputConfigParam struct { + Voice param.Opt[string] `json:"voice,omitzero"` + // contains filtered or unexported fields +} +``` + + +### func \(RealtimeAudioOutputConfigParam\) [MarshalJSON]() + +```go +func (r RealtimeAudioOutputConfigParam) MarshalJSON() ([]byte, error) +``` + + + + +## type [RealtimeAudioPayload]() + + + +```go +type RealtimeAudioPayload struct { + Audio string `json:"audio"` +} +``` + + +## type [RealtimeCapabilities]() + + + +```go +type RealtimeCapabilities struct { + GracefulClose bool `json:"graceful_close"` +} +``` + + +## type [RealtimeClientEvent]() + +RealtimeClientEvent is always encoded as a JSON text frame. Use SendEvent for protocol extensions; convenience methods provide validated public events. + +```go +type RealtimeClientEvent struct { + Version string `json:"version"` + Type string `json:"type"` + Payload map[string]any `json:"payload"` +} +``` + + +## type [RealtimeClosedPayload]() + + + +```go +type RealtimeClosedPayload struct { + RequestID string `json:"request_id"` + Outcome string `json:"outcome"` +} +``` + + +## type [RealtimeConfig]() + + + +```go +type RealtimeConfig struct { + Audio RealtimeAudioConfig `json:"audio"` +} +``` + + +## type [RealtimeConfigParam]() + + + +```go +type RealtimeConfigParam struct { + Audio RealtimeAudioConfigParam `json:"audio,omitzero"` + // contains filtered or unexported fields +} +``` + + +### func \(RealtimeConfigParam\) [MarshalJSON]() + +```go +func (r RealtimeConfigParam) MarshalJSON() ([]byte, error) +``` + + + + +## type [RealtimeConnectParams]() + +RealtimeConnectParams controls per\-stage timeouts and bounded buffering. Zero values select defaults; negative values are rejected. The caller's ctx spans the connection lifetime, independently of ordinary HTTP timeouts. + +```go +type RealtimeConnectParams struct { + HandshakeTimeout time.Duration // Default 30s. + ReadyTimeout time.Duration // Default 60s. + SendTimeout time.Duration // Default 30s. + CloseTimeout time.Duration // Default 10s. + QueueCapacity int // Default 256 events. + QueueBytes int64 // Default 4 MiB. + ReadLimit int64 // Default 1 MiB per server message. +} +``` + + +## type [RealtimeConnection]() + +RealtimeConnection has one receiver and serializes sends. Ready is immutable after Connect returns. Applications own audio playback and its receipts. + +```go +type RealtimeConnection struct { + ConversationID string + RequestID string + Ready RealtimeReadyPayload + // contains filtered or unexported fields +} +``` + + +### func \(\*RealtimeConnection\) [AppendAudio]() + +```go +func (c *RealtimeConnection) AppendAudio(ctx context.Context, pcm []byte) error +``` + +AppendAudio sends nonempty 16 kHz mono PCM16 little\-endian bytes. It does not resample, decode WAV or send binary WebSocket frames. + + +### func \(\*RealtimeConnection\) [Close]() + +```go +func (c *RealtimeConnection) Close() error +``` + +Close releases transport resources without confirming history persistence. + + +### func \(\*RealtimeConnection\) [CloseGracefully]() + +```go +func (c *RealtimeConnection) CloseGracefully(ctx context.Context) (*RealtimeClosedPayload, error) +``` + +CloseGracefully sends a unique close request, waits for matching persistence acknowledgement, then closes the transport. Ordinary Close does not confirm persistence. Repeated successful calls return the same acknowledgement. + + +### func \(\*RealtimeConnection\) [Err]() + +```go +func (c *RealtimeConnection) Err() error +``` + +Err returns the terminal receive/transport error, if any. + + +### func \(\*RealtimeConnection\) [Interrupt]() + +```go +func (c *RealtimeConnection) Interrupt(ctx context.Context) error +``` + + + + +### func \(\*RealtimeConnection\) [Ping]() + +```go +func (c *RealtimeConnection) Ping(ctx context.Context) error +``` + + + + +### func \(\*RealtimeConnection\) [PlaybackCancelled]() + +```go +func (c *RealtimeConnection) PlaybackCancelled(ctx context.Context, r RealtimePlaybackReceipt) error +``` + + + + +### func \(\*RealtimeConnection\) [PlaybackEnded]() + +```go +func (c *RealtimeConnection) PlaybackEnded(ctx context.Context, r RealtimePlaybackReceipt) error +``` + + + + +### func \(\*RealtimeConnection\) [PlaybackStarted]() + +```go +func (c *RealtimeConnection) PlaybackStarted(ctx context.Context, r RealtimePlaybackReceipt) error +``` + + + + +### func \(\*RealtimeConnection\) [Recv]() + +```go +func (c *RealtimeConnection) Recv(ctx context.Context) (RealtimeEvent, error) +``` + +Recv preserves receive order and drains buffered events before reporting a terminal error. Cancelling this wait does not close the connection. + + +### func \(\*RealtimeConnection\) [SendEvent]() + +```go +func (c *RealtimeConnection) SendEvent(ctx context.Context, event RealtimeClientEvent) error +``` + + + + +### func \(\*RealtimeConnection\) [SendText]() + +```go +func (c *RealtimeConnection) SendText(ctx context.Context, text string) error +``` + + + + +## type [RealtimeConversation]() + +RealtimeConversation timestamps remain text to preserve the wire precision. + +```go +type RealtimeConversation struct { + ID string `json:"id"` + Type string `json:"type"` + Status string `json:"status"` + Title *string `json:"title"` + Metadata map[string]any `json:"metadata"` + Config RealtimeConfig `json:"config"` + CreatedAt string `json:"created_at"` + UpdatedAt string `json:"updated_at"` +} +``` + + +## type [RealtimeConversationHistoryParams]() + + + +```go +type RealtimeConversationHistoryParams struct { + Limit param.Opt[int64] `query:"limit,omitzero" json:"-"` + Before param.Opt[string] `query:"before,omitzero" json:"-"` + // "message", "work", or "message,work"; the default includes both. + Types param.Opt[string] `query:"types,omitzero" json:"-"` + // contains filtered or unexported fields +} +``` + + +### func \(RealtimeConversationHistoryParams\) [URLQuery]() + +```go +func (r RealtimeConversationHistoryParams) URLQuery() (url.Values, error) +``` + + + + +## type [RealtimeConversationNewParams]() + + + +```go +type RealtimeConversationNewParams struct { + IdentityID string `json:"identity_id" api:"required"` + TemplateID string `json:"template_id" api:"required"` + Title param.Opt[string] `json:"title,omitzero"` + Metadata map[string]any `json:"metadata,omitzero"` + Config RealtimeConfigParam `json:"config,omitzero"` + IdempotencyKey param.Opt[string] `header:"Idempotency-Key,omitzero" json:"-"` + // contains filtered or unexported fields +} +``` + + +### func \(RealtimeConversationNewParams\) [MarshalJSON]() + +```go +func (r RealtimeConversationNewParams) MarshalJSON() ([]byte, error) +``` + + + + +## type [RealtimeConversationService]() + +RealtimeConversationService creates Realtime Conversations and queries their persisted history. Realtime is Beta; a Template must already have it enabled. + +```go +type RealtimeConversationService struct{ Options []option.RequestOption } +``` + + +### func [NewRealtimeConversationService]() + +```go +func NewRealtimeConversationService(opts ...option.RequestOption) RealtimeConversationService +``` + + + + +### func \(\*RealtimeConversationService\) [History]() + +```go +func (r *RealtimeConversationService) History(ctx context.Context, conversationID string, params RealtimeConversationHistoryParams, opts ...option.RequestOption) (res *RealtimeHistoryPage, err error) +``` + +History fetches a page of persisted events. Its cursor points towards older history, and is unrelated to WebSocket event IDs or sequence numbers. + + +### func \(\*RealtimeConversationService\) [HistoryAutoPaging]() + +```go +func (r *RealtimeConversationService) HistoryAutoPaging(ctx context.Context, conversationID string, params RealtimeConversationHistoryParams, opts ...option.RequestOption) *RealtimeHistoryAutoPager +``` + +HistoryAutoPaging iterates persisted history towards older records. + + +### func \(\*RealtimeConversationService\) [New]() + +```go +func (r *RealtimeConversationService) New(ctx context.Context, params RealtimeConversationNewParams, opts ...option.RequestOption) (res *RealtimeConversation, err error) +``` + +New creates a Conversation. Supply a stable Idempotency\-Key for this operation with IdempotencyKey or option.WithHeader; it is required by the service. + + +## type [RealtimeErrorPayload]() + + + +```go +type RealtimeErrorPayload struct { + Code string `json:"code"` + Message string `json:"message"` + Retryable *bool `json:"retryable,omitempty"` + RetryAfterMS *int64 `json:"retry_after_ms,omitempty"` +} +``` + + +## type [RealtimeEvent]() + +RealtimeEvent is a server event. Payload and RawJSON preserve Beta extension fields; Timestamp preserves nanosecond precision. Sequence is not a cursor. + +```go +type RealtimeEvent struct { + Version string `json:"version"` + Type string `json:"type"` + EventID string `json:"event_id"` + Sequence int64 `json:"sequence"` + ConversationID string `json:"conversation_id"` + Timestamp string `json:"timestamp"` + WorkID string `json:"work_id,omitempty"` + AnnouncementID string `json:"announcement_id,omitempty"` + Payload json.RawMessage `json:"payload"` + // contains filtered or unexported fields +} +``` + + +### func \(RealtimeEvent\) [Audio]() + +```go +func (e RealtimeEvent) Audio() ([]byte, error) +``` + +Audio decodes audio.delta PCM16. It does not decode WAV or play audio. + + +### func \(RealtimeEvent\) [DecodePayload]() + +```go +func (e RealtimeEvent) DecodePayload(dst any) error +``` + +DecodePayload decodes a known typed payload or an application's extension. + + +### func \(RealtimeEvent\) [RawJSON]() + +```go +func (e RealtimeEvent) RawJSON() string +``` + + + + +## type [RealtimeHistoryAutoPager]() + + + +```go +type RealtimeHistoryAutoPager struct { + // contains filtered or unexported fields +} +``` + + +### func \(\*RealtimeHistoryAutoPager\) [Current]() + +```go +func (p *RealtimeHistoryAutoPager) Current() RealtimeHistoryEvent +``` + + + + +### func \(\*RealtimeHistoryAutoPager\) [Err]() + +```go +func (p *RealtimeHistoryAutoPager) Err() error +``` + + + + +### func \(\*RealtimeHistoryAutoPager\) [Next]() + +```go +func (p *RealtimeHistoryAutoPager) Next() bool +``` + + + + +## type [RealtimeHistoryConversation]() + + + +```go +type RealtimeHistoryConversation struct { + ID string `json:"id"` + Title *string `json:"title"` + InitializationStatus string `json:"initialization_status"` + Metadata map[string]any `json:"metadata"` + CreatedAt string `json:"created_at"` + UpdatedAt string `json:"updated_at"` +} +``` + + +## type [RealtimeHistoryEvent]() + +RealtimeHistoryEvent retains unknown fields through RawJSON. ID is a history ID \(vevt\_\*\), not an ephemeral WebSocket event\_id. + +```go +type RealtimeHistoryEvent struct { + ID string `json:"id"` + Type string `json:"type"` + Status string `json:"status"` + OccurredAt string `json:"occurred_at"` + Role string `json:"role,omitempty"` + Text string `json:"text,omitempty"` + WorkID string `json:"work_id,omitempty"` + Objective string `json:"objective,omitempty"` + Result string `json:"result,omitempty"` + Error *RealtimeWorkError `json:"error,omitempty"` + TurnID string `json:"turn_id,omitempty"` + UserMessageEventID string `json:"user_message_event_id,omitempty"` + AnnouncementID string `json:"announcement_id,omitempty"` + MilestoneType string `json:"milestone_type,omitempty"` + // contains filtered or unexported fields +} +``` + + +### func \(RealtimeHistoryEvent\) [RawJSON]() + +```go +func (e RealtimeHistoryEvent) RawJSON() string +``` + + + + +### func \(\*RealtimeHistoryEvent\) [UnmarshalJSON]() + +```go +func (e *RealtimeHistoryEvent) UnmarshalJSON(b []byte) error +``` + + + + +## type [RealtimeHistoryPage]() + + + +```go +type RealtimeHistoryPage struct { + Conversation RealtimeHistoryConversation `json:"conversation"` + Events []RealtimeHistoryEvent `json:"events"` + Page RealtimeHistoryPageInfo `json:"page"` + // contains filtered or unexported fields +} +``` + + +### func \(\*RealtimeHistoryPage\) [GetNextPage]() + +```go +func (p *RealtimeHistoryPage) GetNextPage(ctx context.Context) (*RealtimeHistoryPage, error) +``` + +GetNextPage returns older history, or nil at the end. + + +### func \(\*RealtimeHistoryPage\) [HasNextPage]() + +```go +func (p *RealtimeHistoryPage) HasNextPage() bool +``` + + + + +## type [RealtimeHistoryPageInfo]() + + + +```go +type RealtimeHistoryPageInfo struct { + NextBefore *string `json:"next_before"` + HasMore bool `json:"has_more"` +} +``` + + +## type [RealtimePlaybackReceipt]() + + + +```go +type RealtimePlaybackReceipt struct { + WorkID string `json:"work_id,omitempty"` + AnnouncementID string `json:"announcement_id,omitempty"` +} +``` + + +## type [RealtimeProtocolError]() + + + +```go +type RealtimeProtocolError struct{ Message string } +``` + + +### func \(\*RealtimeProtocolError\) [Error]() + +```go +func (e *RealtimeProtocolError) Error() string +``` + + + + +## type [RealtimeReadyPayload]() + + + +```go +type RealtimeReadyPayload struct { + Config RealtimeConfig `json:"config"` + Provider string `json:"provider"` + InputAudio RealtimeAudioFormat `json:"input_audio"` + OutputAudio RealtimeAudioFormat `json:"output_audio"` + Capabilities RealtimeCapabilities `json:"capabilities"` +} +``` + + +## type [RealtimeReasonPayload]() + + + +```go +type RealtimeReasonPayload struct { + Reason string `json:"reason"` +} +``` + + +## type [RealtimeServerError]() + + + +```go +type RealtimeServerError struct { + RealtimeErrorPayload + ConversationID string +} +``` + + +### func \(\*RealtimeServerError\) [Error]() + +```go +func (e *RealtimeServerError) Error() string +``` + + + + +## type [RealtimeService]() + + + +```go +type RealtimeService struct { + Options []option.RequestOption + Conversations RealtimeConversationService +} +``` + + +### func [NewRealtimeService]() + +```go +func NewRealtimeService(opts ...option.RequestOption) RealtimeService +``` + + + + +### func \(\*RealtimeService\) [Connect]() + +```go +func (r *RealtimeService) Connect(ctx context.Context, conversationID string, params RealtimeConnectParams, opts ...option.RequestOption) (*RealtimeConnection, error) +``` + +Connect upgrades once and waits for a validated voice.ready. It never follows redirects or automatically reconnects/replays input. Events received while waiting for ready, including ready itself, remain available through Recv. + + +## type [RealtimeStatePayload]() + + + +```go +type RealtimeStatePayload struct { + State string `json:"state"` +} +``` + + +## type [RealtimeTranscriptPayload]() + + + +```go +type RealtimeTranscriptPayload struct { + Role string `json:"role"` + Text string `json:"text"` + ItemID string `json:"item_id,omitempty"` + ResponseID string `json:"response_id,omitempty"` +} +``` + + +## type [RealtimeTransportError]() + + + +```go +type RealtimeTransportError struct { + ConversationID string + CloseCode int + Reason string + Err error +} +``` + + +### func \(\*RealtimeTransportError\) [Error]() + +```go +func (e *RealtimeTransportError) Error() string +``` + + + + +### func \(\*RealtimeTransportError\) [Unwrap]() + +```go +func (e *RealtimeTransportError) Unwrap() error +``` + + + + +## type [RealtimeWorkAcceptedPayload]() + + + +```go +type RealtimeWorkAcceptedPayload struct { + Status string `json:"status"` + Objective string `json:"objective"` +} +``` + + +## type [RealtimeWorkCompletedPayload]() + + + +```go +type RealtimeWorkCompletedPayload struct { + Status string `json:"status"` + Result string `json:"result"` + EventID string `json:"event_id,omitempty"` +} +``` + + +## type [RealtimeWorkError]() + + + +```go +type RealtimeWorkError struct { + Code string `json:"code"` +} +``` + + +## type [RealtimeWorkFailedPayload]() + + + +```go +type RealtimeWorkFailedPayload struct { + Status string `json:"status"` + Error RealtimeWorkError `json:"error"` + EventID string `json:"event_id,omitempty"` +} +``` + + +## type [RealtimeWorkMilestonePayload]() + + + +```go +type RealtimeWorkMilestonePayload struct { + AnnouncementID string `json:"announcement_id"` + MilestoneType string `json:"milestone_type"` + Summary string `json:"summary"` +} +``` + + +## type [RealtimeWorkProgressPayload]() + + + +```go +type RealtimeWorkProgressPayload struct { + Kind string `json:"kind"` + Detail string `json:"detail"` + Title string `json:"title,omitempty"` + IsError bool `json:"is_error,omitempty"` + EventID string `json:"event_id,omitempty"` +} +``` + + +## type [RealtimeWorkRunningPayload]() + + + +```go +type RealtimeWorkRunningPayload struct { + Status string `json:"status"` +} +``` + ## type [ResourceBinding]() @@ -54978,6 +55930,7 @@ Package convention implements Qoder authentication, transport, request options, - [func \(cfg \*RequestConfig\) Clone\(ctx context.Context\) \*RequestConfig](<#RequestConfig.Clone>) - [func \(cfg \*RequestConfig\) Execute\(\) \(err error\)](<#RequestConfig.Execute>) - [func \(r \*RequestConfig\) MarkAuthorizationDeleted\(\)](<#RequestConfig.MarkAuthorizationDeleted>) + - [func \(cfg \*RequestConfig\) WebSocketHandshake\(\) \(\*http.Client, \*http.Request, error\)](<#RequestConfig.WebSocketHandshake>) - [type RequestOption](<#RequestOption>) - [func EnvironmentAuthentication\(\) RequestOption](<#EnvironmentAuthentication>) - [func RequireCredential\(\) RequestOption](<#RequireCredential>) @@ -55321,6 +56274,15 @@ func (r *RequestConfig) MarkAuthorizationDeleted() MarkAuthorizationDeleted suppresses static/environment header restoration; dynamic credentials retain their existing header\-deletion behavior. + +### func \(\*RequestConfig\) [WebSocketHandshake]() + +```go +func (cfg *RequestConfig) WebSocketHandshake() (*http.Client, *http.Request, error) +``` + +WebSocketHandshake prepares a single upgrade using the same headers, authentication and middleware as HTTP requests. RequestTimeout and the HTTP client's Timeout do not limit the lifetime of the upgraded connection. Custom HTTP doers cannot guarantee WebSocket upgrades; use a native http.Client. + ## type [RequestOption]() diff --git a/examples/forward/realtime/main.go b/examples/forward/realtime/main.go new file mode 100644 index 0000000..3363d30 --- /dev/null +++ b/examples/forward/realtime/main.go @@ -0,0 +1,220 @@ +// Realtime is Beta and requires a pre-enabled Template. This demo uses process +// environment variables; it does not create/delete Identities or Templates. +package main + +import ( + "context" + "encoding/json" + "flag" + "fmt" + "os" + "strings" + "time" + + "github.com/QoderAI/qoder-cloud-agents-sdk-go/convention" + "github.com/QoderAI/qoder-cloud-agents-sdk-go/convention/option" + "github.com/QoderAI/qoder-cloud-agents-sdk-go/examples/internal/live" + "github.com/QoderAI/qoder-cloud-agents-sdk-go/forward" +) + +func main() { + if err := run(); err != nil { + fmt.Fprintln(os.Stderr, "Realtime example failed: "+live.Safe(err)) + os.Exit(1) + } +} + +func run() error { + conversationID := flag.String("conversation", os.Getenv("QODER_CONVERSATION_ID"), "existing Conversation; omit to create one") + identityID := flag.String("identity", os.Getenv("QODER_IDENTITY_ID"), "Identity for a new Conversation") + templateID := flag.String("template", os.Getenv("QODER_TEMPLATE_ID"), "pre-enabled Realtime Template") + idempotencyKey := flag.String("idempotency-key", os.Getenv("QODER_IDEMPOTENCY_KEY"), "stable creation operation key; required when creating") + text := flag.String("text", "", "optional text input (uses model quota)") + pcmFile := flag.String("pcm", "", "optional raw 16 kHz mono PCM16 little-endian input; no WAV header") + historyOnly := flag.Bool("history-only", false, "only paginate an existing Conversation's persisted history") + timeout := flag.Duration("timeout", 2*time.Minute, "overall example timeout") + flag.Parse() + if *text != "" && *pcmFile != "" { + return fmt.Errorf("choose -text or -pcm for this single-turn example") + } + var inputPCM []byte + if *pcmFile != "" { + var err error + inputPCM, err = os.ReadFile(*pcmFile) + if err != nil { + return err + } + if len(inputPCM) == 0 || len(inputPCM)%2 != 0 || len(inputPCM) > 30*16000*2 { + return fmt.Errorf("PCM input must contain complete nonempty 16-bit samples and at most 30 seconds") + } + if len(inputPCM) >= 12 && string(inputPCM[:4]) == "RIFF" && string(inputPCM[8:12]) == "WAVE" { + return fmt.Errorf("PCM input must be raw PCM16, not a WAV file") + } + } + region := os.Getenv("QODER_SA_REGION") + if region == "" { + region = "cn" + } + if region != "cn" && region != "global" { + return fmt.Errorf("QODER_SA_REGION must be cn or global") + } + host := "https://api.qoder.com" + if region == "cn" { + host = "https://api.qoder.com.cn" + } + base := os.Getenv("QODER_FORWARD_BASE_URL") + if base == "" { + base = host + "/api/v1/forward" + } + opts := []option.RequestOption{option.WithBaseURL(base)} + mode := os.Getenv("QODER_AUTH_MODE") + if mode == "" { + mode = "pat" + } + if mode != "pat" && mode != "sa" { + return fmt.Errorf("QODER_AUTH_MODE must be pat or sa") + } + opts = append(opts, option.WithAuthMode(mode)) + if mode == "sa" && os.Getenv("QODER_SA_KEY") != "" { + if os.Getenv("QODER_ACCESS_TOKEN") != "" { + return fmt.Errorf("choose SA Key exchange or direct SAT, not both") + } + credential, err := convention.NewServiceAccountCredential(convention.ServiceAccountCredentialConfig{SAKey: os.Getenv("QODER_SA_KEY"), Region: region, Scopes: []string{"forward.access"}}) + if err != nil { + return err + } + opts = append(opts, option.WithCredential(credential)) + } + client := forward.NewClient(opts...) + ctx, cancel := context.WithTimeout(context.Background(), *timeout) + defer cancel() + if *historyOnly && *conversationID == "" { + return fmt.Errorf("history-only requires -conversation") + } + if *conversationID == "" { + if *identityID == "" || *templateID == "" || *idempotencyKey == "" { + return fmt.Errorf("creation requires -identity, -template and a stable -idempotency-key") + } + conversation, err := client.Realtime.Conversations.New(ctx, forward.RealtimeConversationNewParams{IdentityID: *identityID, TemplateID: *templateID, IdempotencyKey: forward.String(*idempotencyKey), Title: forward.String("SDK Realtime example")}) + if err != nil { + return err + } + *conversationID = conversation.ID + fmt.Println("Conversation:", *conversationID, "(retained by the service; closing does not delete it)") + } + if !*historyOnly { + conn, err := client.Realtime.Connect(ctx, *conversationID, forward.RealtimeConnectParams{}) + if err != nil { + return err + } + defer conn.Close() + fmt.Printf("Ready: input=%d Hz output=%d Hz request_id=%s\n", conn.Ready.InputAudio.SampleRate, conn.Ready.OutputAudio.SampleRate, conn.RequestID) + if *text != "" { + if err = conn.SendText(ctx, *text); err != nil { + return err + } + } + audioErr := make(chan error, 1) + sentAudio := len(inputPCM) == 0 + if len(inputPCM) > 0 { + go func() { + // 640 bytes = 20 ms at 16 kHz, mono PCM16. Pace input in real time. + for start := 0; start < len(inputPCM); start += 640 { + end := min(start+640, len(inputPCM)) + if err := conn.AppendAudio(ctx, inputPCM[start:end]); err != nil { + audioErr <- err + return + } + timer := time.NewTimer(20 * time.Millisecond) + select { + case <-ctx.Done(): + timer.Stop() + audioErr <- ctx.Err() + return + case <-timer.C: + } + } + audioErr <- nil + }() + } + if err = conn.Ping(ctx); err != nil { + return err + } + seenUser, seenAssistant, seenAudioDone := len(inputPCM) == 0, false, false + receipts := map[string]bool{} + for { + select { + case sendErr := <-audioErr: + if sendErr != nil { + return sendErr + } + sentAudio = true + default: + } + event, recvErr := conn.Recv(ctx) + if recvErr != nil { + return recvErr + } + fmt.Printf("Event: %s work_id=%s announcement_id=%s\n", event.Type, event.WorkID, event.AnnouncementID) + if event.Type == "transcript.final" { + var p forward.RealtimeTranscriptPayload + if err = event.DecodePayload(&p); err != nil { + return err + } + if strings.TrimSpace(p.Text) != "" { + seenUser = seenUser || p.Role == "user" + seenAssistant = seenAssistant || p.Role == "assistant" + } + } + if event.Type == "audio.delta" { + pcm, decodeErr := event.Audio() + if decodeErr != nil { + return decodeErr + } + fmt.Println("Output PCM bytes:", len(pcm)) + // This demo has no speaker; never claim playback started/ended. + key := event.WorkID + "/" + event.AnnouncementID + if !receipts[key] { + receipts[key] = true + if err = conn.PlaybackCancelled(ctx, forward.RealtimePlaybackReceipt{WorkID: event.WorkID, AnnouncementID: event.AnnouncementID}); err != nil { + return err + } + } + } + if event.Type == "error" { + var p forward.RealtimeErrorPayload + _ = json.Unmarshal(event.Payload, &p) + return fmt.Errorf("Realtime returned error code %s", p.Code) + } + if event.Type == "audio.done" { + seenAudioDone = true + } + if (event.Type == "pong" && *text == "" && len(inputPCM) == 0) || (seenUser && seenAssistant && seenAudioDone) { + break + } + } + if !sentAudio { + select { + case err = <-audioErr: + if err != nil { + return err + } + case <-ctx.Done(): + return ctx.Err() + } + } + closeCtx, cancelClose := context.WithTimeout(context.Background(), 15*time.Second) + result, closeErr := conn.CloseGracefully(closeCtx) + cancelClose() + if closeErr != nil { + return closeErr + } + fmt.Println("Persisted close outcome:", result.Outcome) + } + pager := client.Realtime.Conversations.HistoryAutoPaging(ctx, *conversationID, forward.RealtimeConversationHistoryParams{Types: forward.String("message,work"), Limit: forward.Int(50)}) + for pager.Next() { + event := pager.Current() + fmt.Printf("History: %s %s\n", event.ID, event.Type) + } + return pager.Err() +} diff --git a/forward/client.go b/forward/client.go index c8ee011..28a9650 100644 --- a/forward/client.go +++ b/forward/client.go @@ -29,6 +29,7 @@ type Client struct { Usage UsageService Models ModelService ServiceAccountTokens ServiceAccountTokenService + Realtime RealtimeService } // NewClient uses common authentication defaults and QODER_FORWARD_BASE_URL. Explicit options win. @@ -59,5 +60,6 @@ func NewClient(opts ...option.RequestOption) Client { Usage: NewUsageService(opts...), Models: NewModelService(opts...), ServiceAccountTokens: NewServiceAccountTokenService(opts...), + Realtime: NewRealtimeService(opts...), } } diff --git a/forward/client_test.go b/forward/client_test.go index 4e175a1..ea67f71 100644 --- a/forward/client_test.go +++ b/forward/client_test.go @@ -26,7 +26,7 @@ func TestForwardAPIInventory(t *testing.T) { for j := 0; j < methods.NumMethod(); j++ { name := methods.Method(j).Name key := field.Type().Name() + "." + name - if !strings.HasSuffix(name, "AutoPaging") && key != "SessionEventService.NewResumableStream" { + if !strings.HasSuffix(name, "AutoPaging") && key != "SessionEventService.NewResumableStream" && key != "RealtimeService.Connect" { actual[key] = true } } @@ -34,8 +34,8 @@ func TestForwardAPIInventory(t *testing.T) { } } walk(root) - if len(actual) != 116 { - t.Fatalf("HTTP methods: %d want 116", len(actual)) + if len(actual) != 118 { + t.Fatalf("HTTP methods: %d want 118", len(actual)) } baseline := map[string]bool{} for _, c := range operations(t) { diff --git a/forward/realtime.go b/forward/realtime.go new file mode 100644 index 0000000..10ffbc6 --- /dev/null +++ b/forward/realtime.go @@ -0,0 +1,493 @@ +package forward + +import ( + "context" + "crypto/rand" + "encoding/base64" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "io" + "net/http" + "slices" + "strings" + "sync" + "time" + "unicode/utf8" + + "github.com/QoderAI/qoder-cloud-agents-sdk-go/convention" + "github.com/QoderAI/qoder-cloud-agents-sdk-go/convention/apierror" + "github.com/QoderAI/qoder-cloud-agents-sdk-go/convention/option" + "github.com/coder/websocket" +) + +const RealtimeMaxClientMessageBytes = 256 * 1024 + +type RealtimeService struct { + Options []option.RequestOption + Conversations RealtimeConversationService +} + +func NewRealtimeService(opts ...option.RequestOption) RealtimeService { + return RealtimeService{Options: slices.Clone(opts), Conversations: NewRealtimeConversationService(opts...)} +} + +// RealtimeConnectParams controls per-stage timeouts and bounded buffering. +// Zero values select defaults; negative values are rejected. The caller's ctx +// spans the connection lifetime, independently of ordinary HTTP timeouts. +type RealtimeConnectParams struct { + HandshakeTimeout time.Duration // Default 30s. + ReadyTimeout time.Duration // Default 60s. + SendTimeout time.Duration // Default 30s. + CloseTimeout time.Duration // Default 10s. + QueueCapacity int // Default 256 events. + QueueBytes int64 // Default 4 MiB. + ReadLimit int64 // Default 1 MiB per server message. +} + +func (p RealtimeConnectParams) defaults() (RealtimeConnectParams, error) { + if p.HandshakeTimeout < 0 || p.ReadyTimeout < 0 || p.SendTimeout < 0 || p.CloseTimeout < 0 || p.QueueCapacity < 0 || p.QueueBytes < 0 || p.ReadLimit < 0 { + return p, fmt.Errorf("Realtime timeouts and limits must not be negative") + } + if p.HandshakeTimeout == 0 { + p.HandshakeTimeout = 30 * time.Second + } + if p.ReadyTimeout == 0 { + p.ReadyTimeout = 60 * time.Second + } + if p.SendTimeout == 0 { + p.SendTimeout = 30 * time.Second + } + if p.CloseTimeout == 0 { + p.CloseTimeout = 10 * time.Second + } + if p.QueueCapacity == 0 { + p.QueueCapacity = 256 + } + if p.QueueBytes == 0 { + p.QueueBytes = 4 * 1024 * 1024 + } + if p.ReadLimit == 0 { + p.ReadLimit = 1024 * 1024 + } + return p, nil +} + +// Connect upgrades once and waits for a validated voice.ready. It never follows +// redirects or automatically reconnects/replays input. Events received while +// waiting for ready, including ready itself, remain available through Recv. +func (r *RealtimeService) Connect(ctx context.Context, conversationID string, params RealtimeConnectParams, opts ...option.RequestOption) (*RealtimeConnection, error) { + if strings.TrimSpace(conversationID) == "" { + return nil, fmt.Errorf("missing required conversation_id parameter") + } + params, err := params.defaults() + if err != nil { + return nil, err + } + handshakeCtx, cancel := context.WithTimeout(ctx, params.HandshakeTimeout) + defer cancel() + cfg, err := convention.NewRequestConfig(handshakeCtx, http.MethodGet, "realtime", nil, nil, slices.Concat(r.Options, opts)...) + if err != nil { + return nil, err + } + client, request, err := cfg.WebSocketHandshake() + if err != nil { + return nil, err + } + query := request.URL.Query() + query.Set("conversation_id", conversationID) + request.URL.RawQuery = query.Encode() + ws, response, err := websocket.Dial(handshakeCtx, request.URL.String(), &websocket.DialOptions{HTTPClient: client, HTTPHeader: request.Header}) + cancel() + if cfg.ResponseInto != nil { + *cfg.ResponseInto = response + } + if err != nil { + if response != nil && response.StatusCode != http.StatusSwitchingProtocols { + api := &apierror.Error{StatusCode: response.StatusCode, Response: response, Request: response.Request, RequestID: response.Header.Get("X-Request-Id")} + if response.Body != nil { + body, _ := io.ReadAll(response.Body) + _ = response.Body.Close() + _ = api.UnmarshalJSON(body) + } + return nil, api + } + return nil, err + } + lifetime, cancelLifetime := context.WithCancel(ctx) + c := &RealtimeConnection{ConversationID: conversationID, ws: ws, params: params, ctx: lifetime, cancel: cancelLifetime, events: make(chan realtimeQueuedEvent, params.QueueCapacity), done: make(chan struct{}), readyDone: make(chan struct{}), sendGate: make(chan struct{}, 1), closeGate: make(chan struct{}, 1)} + if response != nil { + c.RequestID = response.Header.Get("X-Request-Id") + } + ws.SetReadLimit(params.ReadLimit) + go c.receiveLoop() + readyCtx, cancelReady := context.WithTimeout(ctx, params.ReadyTimeout) + defer cancelReady() + select { + case <-c.readyDone: + c.mu.Lock() + isReady := c.ready + c.mu.Unlock() + if !isReady { + return nil, c.Err() + } + return c, nil + case <-c.done: + return nil, c.Err() + case <-readyCtx.Done(): + err := fmt.Errorf("qca realtime ready wait: %w", readyCtx.Err()) + c.finish(err) + return nil, err + } +} + +type realtimeQueuedEvent struct { + event RealtimeEvent + bytes int64 +} + +// RealtimeConnection has one receiver and serializes sends. Ready is immutable +// after Connect returns. Applications own audio playback and its receipts. +type RealtimeConnection struct { + ConversationID string + RequestID string + Ready RealtimeReadyPayload + ws *websocket.Conn + params RealtimeConnectParams + ctx context.Context + cancel context.CancelFunc + events chan realtimeQueuedEvent + done chan struct{} + readyDone chan struct{} + sendGate chan struct{} + closeGate chan struct{} + mu sync.Mutex + ready bool + queuedBytes int64 + err error + closing bool + closeRequestID string + closeResult *RealtimeClosedPayload + closeWaiter chan RealtimeClosedPayload + finishOnce sync.Once +} + +type RealtimeTransportError struct { + ConversationID string + CloseCode int + Reason string + Err error +} + +func (e *RealtimeTransportError) Error() string { + return fmt.Sprintf("qca realtime transport closed (conversation_id=%s, code=%d): %s", e.ConversationID, e.CloseCode, e.Reason) +} +func (e *RealtimeTransportError) Unwrap() error { return e.Err } + +func (c *RealtimeConnection) receiveLoop() { + for { + kind, b, err := c.ws.Read(c.ctx) + if err != nil { + var ce websocket.CloseError + if errors.As(err, &ce) { + c.finish(&RealtimeTransportError{ConversationID: c.ConversationID, CloseCode: int(ce.Code), Reason: ce.Reason, Err: err}) + } else { + c.finish(err) + } + return + } + if kind != websocket.MessageText { + c.finish(&RealtimeProtocolError{Message: "server sent a binary frame"}) + return + } + event, err := decodeRealtimeEvent(b, c.ConversationID) + if err != nil { + c.finish(err) + return + } + var ready *RealtimeReadyPayload + if event.Type == "voice.ready" { + var p RealtimeReadyPayload + var capabilityFields struct { + Capabilities map[string]json.RawMessage `json:"capabilities"` + } + capErr := event.DecodePayload(&capabilityFields) + graceful := capabilityFields.Capabilities["graceful_close"] + if capErr != nil || graceful == nil || string(graceful) == "null" { + c.finish(&RealtimeProtocolError{Message: "voice.ready requires capabilities.graceful_close"}) + return + } + if err := event.DecodePayload(&p); err != nil || p.Provider == "" || p.Config.Audio.Output.Voice == "" || p.InputAudio.Format != "pcm16" || p.InputAudio.SampleRate != 16000 || p.InputAudio.Channels != 1 || p.OutputAudio.Format != "pcm16" || p.OutputAudio.SampleRate != 24000 || p.OutputAudio.Channels != 1 { + c.finish(&RealtimeProtocolError{Message: "invalid voice.ready configuration"}) + return + } + ready = &p + } + c.mu.Lock() + if event.Type == "connection.closed" && c.closeWaiter != nil { + var p RealtimeClosedPayload + if event.DecodePayload(&p) == nil && p.RequestID == c.closeRequestID { + select { + case c.closeWaiter <- p: + default: + } + } + } + if c.queuedBytes+int64(len(b)) > c.params.QueueBytes { + c.mu.Unlock() + c.finish(&RealtimeProtocolError{Message: "receive queue byte limit exceeded"}) + return + } + c.queuedBytes += int64(len(b)) + c.mu.Unlock() + select { + case c.events <- realtimeQueuedEvent{event: event, bytes: int64(len(b))}: + default: + c.finish(&RealtimeProtocolError{Message: "receive queue event limit exceeded"}) + return + } + if ready != nil { + c.mu.Lock() + if !c.ready { + c.Ready = *ready + c.ready = true + close(c.readyDone) + } + c.mu.Unlock() + } + if event.Type == "voice.replaced" { + c.finish(&RealtimeProtocolError{Message: "connection replaced"}) + return + } + if event.Type == "error" { + var p RealtimeErrorPayload + if err := event.DecodePayload(&p); err != nil { + c.finish(&RealtimeProtocolError{Message: "invalid error payload"}) + return + } + c.mu.Lock() + isReady := c.ready + c.mu.Unlock() + if !isReady || p.Code == "provider_initialization_failed" || p.Code == "context_restore_failed" || p.Code == "provider_unavailable" || p.Code == "service_restarting" || p.Code == "event_persistence_failed" { + c.finish(&RealtimeServerError{RealtimeErrorPayload: p, ConversationID: c.ConversationID}) + return + } + } + } +} + +func (c *RealtimeConnection) finish(err error) { + c.finishOnce.Do(func() { + c.mu.Lock() + c.err = err + c.mu.Unlock() + close(c.done) + c.cancel() + _ = c.ws.CloseNow() + }) +} + +// Err returns the terminal receive/transport error, if any. +func (c *RealtimeConnection) Err() error { c.mu.Lock(); defer c.mu.Unlock(); return c.err } + +// Recv preserves receive order and drains buffered events before reporting a +// terminal error. Cancelling this wait does not close the connection. +func (c *RealtimeConnection) Recv(ctx context.Context) (RealtimeEvent, error) { + dequeue := func(q realtimeQueuedEvent) (RealtimeEvent, error) { + c.mu.Lock() + c.queuedBytes -= q.bytes + c.mu.Unlock() + return q.event, nil + } + select { + case q := <-c.events: + return dequeue(q) + default: + } + select { + case q := <-c.events: + return dequeue(q) + case <-ctx.Done(): + return RealtimeEvent{}, ctx.Err() + case <-c.done: + select { + case q := <-c.events: + return dequeue(q) + default: + } + if err := c.Err(); err != nil { + return RealtimeEvent{}, err + } + return RealtimeEvent{}, io.EOF + } +} + +func (c *RealtimeConnection) SendEvent(ctx context.Context, event RealtimeClientEvent) error { + return c.sendEvent(ctx, event, false) +} +func (c *RealtimeConnection) sendEvent(ctx context.Context, event RealtimeClientEvent, allowClosing bool) error { + if event.Version == "" { + event.Version = RealtimeProtocolVersion + } + if event.Version != RealtimeProtocolVersion || event.Type == "" { + return &RealtimeProtocolError{Message: "invalid client event envelope"} + } + if event.Payload == nil { + event.Payload = map[string]any{} + } + b, err := json.Marshal(event) + if err != nil { + return err + } + if len(b) > RealtimeMaxClientMessageBytes { + return &RealtimeProtocolError{Message: "client message exceeds 256 KiB"} + } + sendCtx, cancel := context.WithTimeout(ctx, c.params.SendTimeout) + defer cancel() + select { + case c.sendGate <- struct{}{}: + defer func() { <-c.sendGate }() + case <-sendCtx.Done(): + return sendCtx.Err() + case <-c.done: + if err := c.Err(); err != nil { + return err + } + return io.ErrClosedPipe + } + select { + case <-c.done: + if err := c.Err(); err != nil { + return err + } + return io.ErrClosedPipe + default: + } + c.mu.Lock() + closing := c.closing + c.mu.Unlock() + if closing && !allowClosing { + return &RealtimeProtocolError{Message: "connection is closing"} + } + if err = c.ws.Write(sendCtx, websocket.MessageText, b); err != nil { + c.finish(err) + } + return err +} +func (c *RealtimeConnection) SendText(ctx context.Context, text string) error { + if !utf8.ValidString(text) { + return &RealtimeProtocolError{Message: "text must be valid UTF-8"} + } + text = strings.TrimSpace(text) + if text == "" || len([]byte(text)) > 16*1024 { + return &RealtimeProtocolError{Message: "text must be nonempty valid UTF-8 and at most 16 KiB"} + } + return c.SendEvent(ctx, RealtimeClientEvent{Type: "text.message", Payload: map[string]any{"text": text}}) +} + +// AppendAudio sends nonempty 16 kHz mono PCM16 little-endian bytes. It does not +// resample, decode WAV or send binary WebSocket frames. +func (c *RealtimeConnection) AppendAudio(ctx context.Context, pcm []byte) error { + if len(pcm) == 0 || len(pcm)%2 != 0 { + return &RealtimeProtocolError{Message: "audio must contain whole nonempty PCM16 samples"} + } + if base64.StdEncoding.EncodedLen(len(pcm)) > RealtimeMaxClientMessageBytes { + return &RealtimeProtocolError{Message: "client message exceeds 256 KiB"} + } + return c.SendEvent(ctx, RealtimeClientEvent{Type: "audio.append", Payload: map[string]any{"audio": base64.StdEncoding.EncodeToString(pcm)}}) +} +func (c *RealtimeConnection) Interrupt(ctx context.Context) error { + return c.SendEvent(ctx, RealtimeClientEvent{Type: "interrupt"}) +} +func (c *RealtimeConnection) Ping(ctx context.Context) error { + return c.SendEvent(ctx, RealtimeClientEvent{Type: "ping"}) +} +func (c *RealtimeConnection) playback(ctx context.Context, kind string, receipt RealtimePlaybackReceipt) error { + payload := map[string]any{} + if receipt.WorkID != "" { + payload["work_id"] = receipt.WorkID + } + if receipt.AnnouncementID != "" { + payload["announcement_id"] = receipt.AnnouncementID + } + return c.SendEvent(ctx, RealtimeClientEvent{Type: kind, Payload: payload}) +} +func (c *RealtimeConnection) PlaybackStarted(ctx context.Context, r RealtimePlaybackReceipt) error { + return c.playback(ctx, "playback.started", r) +} +func (c *RealtimeConnection) PlaybackEnded(ctx context.Context, r RealtimePlaybackReceipt) error { + return c.playback(ctx, "playback.ended", r) +} +func (c *RealtimeConnection) PlaybackCancelled(ctx context.Context, r RealtimePlaybackReceipt) error { + return c.playback(ctx, "playback.cancelled", r) +} + +// CloseGracefully sends a unique close request, waits for matching persistence +// acknowledgement, then closes the transport. Ordinary Close does not confirm +// persistence. Repeated successful calls return the same acknowledgement. +func (c *RealtimeConnection) CloseGracefully(ctx context.Context) (*RealtimeClosedPayload, error) { + closeCtx, cancel := context.WithTimeout(ctx, c.params.CloseTimeout) + defer cancel() + select { + case c.closeGate <- struct{}{}: + defer func() { <-c.closeGate }() + case <-closeCtx.Done(): + return nil, closeCtx.Err() + } + c.mu.Lock() + if c.closeResult != nil { + res := *c.closeResult + c.mu.Unlock() + return &res, nil + } + c.mu.Unlock() + defer c.Close() + if !c.Ready.Capabilities.GracefulClose { + return nil, &RealtimeProtocolError{Message: "server does not support graceful close"} + } + var id [16]byte + if _, err := rand.Read(id[:]); err != nil { + return nil, err + } + requestID := hex.EncodeToString(id[:]) + c.mu.Lock() + c.closing = true + c.closeRequestID = requestID + c.closeWaiter = make(chan RealtimeClosedPayload, 1) + waiter := c.closeWaiter + c.mu.Unlock() + if err := c.sendEvent(closeCtx, RealtimeClientEvent{Type: "connection.close", Payload: map[string]any{"request_id": requestID}}, true); err != nil { + return nil, err + } + select { + case result := <-waiter: + if result.Outcome != "no_active_response" && result.Outcome != "no_content" && result.Outcome != "saved" && result.Outcome != "saved_interrupted" { + return nil, &RealtimeProtocolError{Message: "invalid connection.closed outcome"} + } + c.mu.Lock() + c.closeResult = &result + c.mu.Unlock() + return &result, nil + case <-closeCtx.Done(): + return nil, fmt.Errorf("qca realtime close acknowledgement: %w", closeCtx.Err()) + case <-c.done: + select { + case result := <-waiter: + if result.Outcome == "no_active_response" || result.Outcome == "no_content" || result.Outcome == "saved" || result.Outcome == "saved_interrupted" { + c.mu.Lock() + c.closeResult = &result + c.mu.Unlock() + return &result, nil + } + default: + } + if err := c.Err(); err != nil { + return nil, err + } + return nil, io.ErrClosedPipe + } +} + +// Close releases transport resources without confirming history persistence. +func (c *RealtimeConnection) Close() error { c.finish(nil); return nil } diff --git a/forward/realtime_test.go b/forward/realtime_test.go new file mode 100644 index 0000000..daceeaa --- /dev/null +++ b/forward/realtime_test.go @@ -0,0 +1,624 @@ +package forward_test + +import ( + "context" + "encoding/json" + "errors" + "io" + "net/http" + "net/http/httptest" + "reflect" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/QoderAI/qoder-cloud-agents-sdk-go/convention" + "github.com/QoderAI/qoder-cloud-agents-sdk-go/convention/option" + "github.com/QoderAI/qoder-cloud-agents-sdk-go/forward" + "github.com/coder/websocket" +) + +func realtimeFixtures(t *testing.T) []map[string]any { + t.Helper() + var fixture struct { + Events []map[string]any `json:"events"` + } + readJSON(t, "testdata/realtime-events.json", &fixture) + return fixture.Events +} +func writeRealtime(t *testing.T, ws *websocket.Conn, event map[string]any) { + t.Helper() + b, err := json.Marshal(event) + if err != nil { + t.Error(err) + return + } + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + if err := ws.Write(ctx, websocket.MessageText, b); err != nil { + t.Error(err) + } +} +func readRealtime(t *testing.T, ws *websocket.Conn) map[string]any { + t.Helper() + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + kind, b, err := ws.Read(ctx) + if err != nil { + t.Error(err) + return nil + } + if kind != websocket.MessageText { + t.Error("client sent binary frame") + } + var event map[string]any + if err = json.Unmarshal(b, &event); err != nil { + t.Error(err) + } + return event +} +func realtimeServer(t *testing.T, handle func(*websocket.Conn, *http.Request)) *httptest.Server { + t.Helper() + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + ws, err := websocket.Accept(w, r, nil) + if err != nil { + t.Error(err) + return + } + defer ws.CloseNow() + handle(ws, r) + })) + t.Cleanup(server.Close) + return server +} +func realtimeClient(url string, opts ...option.RequestOption) forward.Client { + return forward.NewClient(append([]option.RequestOption{option.WithPAT("test-pat"), option.WithBaseURL(url + "/prefix/api/v1/forward/")}, opts...)...) +} +func realtimeConnect(t *testing.T, client forward.Client, params forward.RealtimeConnectParams) *forward.RealtimeConnection { + t.Helper() + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + t.Cleanup(cancel) + conn, err := client.Realtime.Connect(ctx, "conv_test", params) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = conn.Close() }) + return conn +} + +func TestRealtimeProtocolAndConvenienceMethods(t *testing.T) { + fixture := realtimeFixtures(t) + sent := make(chan []map[string]any, 1) + server := realtimeServer(t, func(ws *websocket.Conn, r *http.Request) { + if r.URL.Path != "/prefix/api/v1/forward/realtime" || r.URL.Query().Get("conversation_id") != "conv_test" || r.Header.Get("Authorization") != "Bearer test-pat" || r.Header.Get("X-Request-Header") != "request" { + t.Error("handshake did not preserve URL/auth/options") + } + for _, event := range fixture[:2] { + writeRealtime(t, ws, event) + } + var got []map[string]any + for i := 0; i < 9; i++ { + got = append(got, readRealtime(t, ws)) + } + sent <- got + for _, event := range fixture[2:16] { + writeRealtime(t, ws, event) + } + closeEvent := readRealtime(t, ws) + requestID := closeEvent["payload"].(map[string]any)["request_id"] + wrong := fixture[16] + wrong["payload"] = map[string]any{"request_id": "wrong", "outcome": "saved"} + writeRealtime(t, ws, wrong) + good := fixture[16] + good["payload"] = map[string]any{"request_id": requestID, "outcome": "saved_interrupted"} + writeRealtime(t, ws, good) + _, _, _ = ws.Read(context.Background()) + }) + client := realtimeClient(server.URL, option.WithHeader("X-Request-Header", "client")) + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + conn, err := client.Realtime.Connect(ctx, "conv_test", forward.RealtimeConnectParams{}, option.WithHeader("X-Request-Header", "request")) + if err != nil { + t.Fatal(err) + } + defer conn.Close() + if conn.Ready.InputAudio.SampleRate != 16000 || conn.Ready.OutputAudio.SampleRate != 24000 { + t.Fatal("ready configuration missing") + } + for i, kind := range []string{"voice.state", "voice.ready"} { + event, err := conn.Recv(ctx) + if err != nil || event.Type != kind || event.Sequence != int64(i+1) { + t.Fatalf("startup events lost: %#v %v", event, err) + } + } + receipt := forward.RealtimePlaybackReceipt{WorkID: "work_one", AnnouncementID: "ann_one"} + for _, send := range []func() error{ + func() error { return conn.SendText(ctx, "你好") }, func() error { return conn.AppendAudio(ctx, []byte{0, 0, 1, 0}) }, func() error { return conn.Interrupt(ctx) }, func() error { return conn.PlaybackStarted(ctx, receipt) }, func() error { return conn.PlaybackEnded(ctx, receipt) }, func() error { return conn.PlaybackCancelled(ctx, receipt) }, func() error { return conn.Ping(ctx) }, func() error { + return conn.SendEvent(ctx, forward.RealtimeClientEvent{Type: "future.client", Payload: map[string]any{"extra": true}}) + }, func() error { return conn.PlaybackEnded(ctx, forward.RealtimePlaybackReceipt{}) }, + } { + if err := send(); err != nil { + t.Fatal(err) + } + } + got := <-sent + for i, kind := range []string{"text.message", "audio.append", "interrupt", "playback.started", "playback.ended", "playback.cancelled", "ping", "future.client", "playback.ended"} { + if got[i]["version"] != "voice.realtime.v1" || got[i]["type"] != kind || got[i]["payload"] == nil { + t.Errorf("bad client event %#v", got[i]) + } + } + if got[0]["payload"].(map[string]any)["text"] != "你好" || got[1]["payload"].(map[string]any)["audio"] != "AAABAA==" || !reflect.DeepEqual(got[3]["payload"], map[string]any{"work_id": "work_one", "announcement_id": "ann_one"}) || len(got[8]["payload"].(map[string]any)) != 0 { + t.Fatalf("client wire payloads %#v", got) + } + for _, expected := range fixture[2:16] { + event, err := conn.Recv(ctx) + if err != nil || event.Type != expected["type"] || !strings.Contains(event.RawJSON(), `"future_envelope":"preserved"`) || event.Timestamp != "2026-10-13T01:02:03.123456789Z" { + t.Fatalf("bad server event %#v %v", event, err) + } + if event.Type == "audio.delta" { + pcm, err := event.Audio() + if err != nil || !reflect.DeepEqual(pcm, []byte{0, 0, 1, 0}) { + t.Fatal("audio decode failed") + } + } + if event.Type == "work.failed" { + var p forward.RealtimeWorkFailedPayload + if event.DecodePayload(&p) != nil || p.Error.Code != "task_error" { + t.Fatal("typed payload decoding failed") + } + } + } + result, err := conn.CloseGracefully(ctx) + if err != nil || result.Outcome != "saved_interrupted" { + t.Fatalf("close failed %#v %v", result, err) + } + again, err := conn.CloseGracefully(ctx) + if err != nil || *again != *result { + t.Fatal("repeated graceful close did not retain confirmation") + } + if conn.SendText(ctx, "after") == nil { + t.Fatal("send after close succeeded") + } +} + +func TestRealtimeInvalidInputBounds(t *testing.T) { + fixture := realtimeFixtures(t) + server := realtimeServer(t, func(ws *websocket.Conn, _ *http.Request) { + writeRealtime(t, ws, fixture[1]) + _, _, _ = ws.Read(context.Background()) + }) + conn := realtimeConnect(t, realtimeClient(server.URL), forward.RealtimeConnectParams{}) + ctx := context.Background() + for _, text := range []string{"", " \t", strings.Repeat("你", 5462), string([]byte{0xff})} { + if conn.SendText(ctx, text) == nil { + t.Fatal("invalid text accepted") + } + } + for _, pcm := range [][]byte{nil, {1}, make([]byte, 256*1024)} { + if conn.AppendAudio(ctx, pcm) == nil { + t.Fatal("invalid audio accepted") + } + } + if conn.SendEvent(ctx, forward.RealtimeClientEvent{Type: "future", Payload: map[string]any{"text": strings.Repeat("x", 256*1024)}}) == nil { + t.Fatal("oversized JSON accepted") + } + if conn.SendEvent(ctx, forward.RealtimeClientEvent{Version: "bad", Type: "ping"}) == nil { + t.Fatal("bad client version accepted") + } +} + +func TestRealtimeReadyCancellationAndHTTPTimeoutIndependence(t *testing.T) { + fixture := realtimeFixtures(t) + server := realtimeServer(t, func(ws *websocket.Conn, _ *http.Request) { + time.Sleep(50 * time.Millisecond) + writeRealtime(t, ws, fixture[1]) + _, _, _ = ws.Read(context.Background()) + }) + client := realtimeClient(server.URL, option.WithHTTPClient(&http.Client{Timeout: time.Millisecond}), option.WithRequestTimeout(time.Millisecond)) + conn := realtimeConnect(t, client, forward.RealtimeConnectParams{}) + ctx, cancel := context.WithCancel(context.Background()) + cancel() + _, _ = conn.Recv(ctx) // queued ready remains observable even with a cancelled wait. + _, err := conn.Recv(ctx) + if !errors.Is(err, context.Canceled) { + t.Fatalf("recv cancellation %v", err) + } + blocked := realtimeServer(t, func(ws *websocket.Conn, _ *http.Request) { _, _, _ = ws.Read(context.Background()) }) + client = realtimeClient(blocked.URL) + _, err = client.Realtime.Connect(context.Background(), "conv_test", forward.RealtimeConnectParams{ReadyTimeout: 10 * time.Millisecond}) + if !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("ready timeout %v", err) + } + ctx, cancel = context.WithCancel(context.Background()) + time.AfterFunc(10*time.Millisecond, cancel) + _, err = client.Realtime.Connect(ctx, "conv_test", forward.RealtimeConnectParams{}) + if !errors.Is(err, context.Canceled) { + t.Fatalf("connect cancellation %v", err) + } +} + +func TestRealtimeProtocolFailures(t *testing.T) { + fixture := realtimeFixtures(t) + for _, kind := range []string{"invalid_json", "binary", "version", "missing_envelope", "null_sequence", "timestamp_comma", "timestamp_offset", "payload_type", "ready_audio", "ready_capability_missing", "ready_capability_type", "initialization_error"} { + t.Run(kind, func(t *testing.T) { + server := realtimeServer(t, func(ws *websocket.Conn, _ *http.Request) { + event := realtimeFixtures(t)[1] + switch kind { + case "invalid_json": + _ = ws.Write(context.Background(), websocket.MessageText, []byte("{broken")) + case "binary": + _ = ws.Write(context.Background(), websocket.MessageBinary, []byte{0, 0}) + case "version": + event["version"] = "v2" + writeRealtime(t, ws, event) + case "missing_envelope": + delete(event, "sequence") + writeRealtime(t, ws, event) + case "null_sequence": + event["sequence"] = nil + writeRealtime(t, ws, event) + case "timestamp_comma": + event["timestamp"] = "2026-10-13T01:02:03,123Z" + writeRealtime(t, ws, event) + case "timestamp_offset": + event["timestamp"] = "2026-10-13T01:02:03+24:00" + writeRealtime(t, ws, event) + case "payload_type": + event["payload"] = []any{} + writeRealtime(t, ws, event) + case "ready_audio": + event["payload"].(map[string]any)["input_audio"] = map[string]any{"format": "wav", "sample_rate": 16000, "channels": 1} + writeRealtime(t, ws, event) + case "ready_capability_missing": + event["payload"].(map[string]any)["capabilities"] = map[string]any{} + writeRealtime(t, ws, event) + case "ready_capability_type": + event["payload"].(map[string]any)["capabilities"] = map[string]any{"graceful_close": "true"} + writeRealtime(t, ws, event) + case "initialization_error": + event["type"] = "error" + event["payload"] = map[string]any{"code": "provider_initialization_failed", "message": "init failed", "retry_after_ms": 5} + writeRealtime(t, ws, event) + } + _, _, _ = ws.Read(context.Background()) + }) + client := realtimeClient(server.URL) + _, err := client.Realtime.Connect(context.Background(), "conv_test", forward.RealtimeConnectParams{ReadyTimeout: time.Second}) + if err == nil { + t.Fatal("bad server protocol accepted") + } + if kind == "initialization_error" { + var serverErr *forward.RealtimeServerError + if !errors.As(err, &serverErr) || serverErr.Code != "provider_initialization_failed" { + t.Fatalf("lost server error %v", err) + } + } else { + var protocol *forward.RealtimeProtocolError + if !errors.As(err, &protocol) { + t.Fatalf("bad error type %T %v", err, err) + } + } + }) + } + release := make(chan struct{}) + server := realtimeServer(t, func(ws *websocket.Conn, _ *http.Request) { + writeRealtime(t, ws, fixture[1]) + <-release + writeRealtime(t, ws, fixture[17]) + _, _, _ = ws.Read(context.Background()) + }) + client := realtimeClient(server.URL) + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + conn, err := client.Realtime.Connect(ctx, "conv_test", forward.RealtimeConnectParams{}) + if err != nil { + t.Fatal(err) + } + close(release) + defer conn.Close() + _, _ = conn.Recv(ctx) + event, err := conn.Recv(ctx) + if err != nil || event.Type != "voice.replaced" { + t.Fatalf("replacement event missing %v", err) + } + _, err = conn.Recv(ctx) + if err == nil { + t.Fatal("replaced connection remained active") + } +} + +func TestRealtimeGracefulCloseFailure(t *testing.T) { + for _, kind := range []string{"wrong_request_id", "persistence_failed", "bad_outcome"} { + t.Run(kind, func(t *testing.T) { + fixture := realtimeFixtures(t) + server := realtimeServer(t, func(ws *websocket.Conn, _ *http.Request) { + writeRealtime(t, ws, fixture[1]) + event := readRealtime(t, ws) + id := event["payload"].(map[string]any)["request_id"] + response := fixture[16] + response["payload"] = map[string]any{"request_id": id, "outcome": "saved"} + switch kind { + case "wrong_request_id": + response["payload"].(map[string]any)["request_id"] = "wrong" + case "persistence_failed": + response["type"] = "error" + response["payload"] = map[string]any{"code": "event_persistence_failed", "message": "storage failed"} + case "bad_outcome": + response["payload"].(map[string]any)["outcome"] = "lost" + } + writeRealtime(t, ws, response) + _, _, _ = ws.Read(context.Background()) + }) + conn := realtimeConnect(t, realtimeClient(server.URL), forward.RealtimeConnectParams{CloseTimeout: 30 * time.Millisecond}) + _, err := conn.CloseGracefully(context.Background()) + if err == nil { + t.Fatal("unconfirmed persistence reported success") + } + if conn.SendText(context.Background(), "closed") == nil { + t.Fatal("failed graceful close leaked active connection") + } + }) + } +} + +func TestRealtimeBackpressureAndReceiveLimit(t *testing.T) { + for _, kind := range []string{"events", "bytes", "message"} { + t.Run(kind, func(t *testing.T) { + fixture := realtimeFixtures(t) + release := make(chan struct{}) + server := realtimeServer(t, func(ws *websocket.Conn, _ *http.Request) { + writeRealtime(t, ws, fixture[1]) + <-release + for i := 0; i < 4; i++ { + event := fixture[15] + size := 400 + if kind == "message" { + size = 800 + } + event["payload"] = map[string]any{"text": strings.Repeat("x", size)} + b, _ := json.Marshal(event) + if ws.Write(context.Background(), websocket.MessageText, b) != nil { + return + } + } + _, _, _ = ws.Read(context.Background()) + }) + params := forward.RealtimeConnectParams{} + switch kind { + case "events": + params.QueueCapacity = 1 + case "bytes": + params.QueueBytes = 900 + case "message": + params.ReadLimit = 650 + } + conn := realtimeConnect(t, realtimeClient(server.URL), params) + _, _ = conn.Recv(context.Background()) + close(release) + deadline := time.Now().Add(time.Second) + for conn.Err() == nil && time.Now().Before(deadline) { + time.Sleep(time.Millisecond) + } + if conn.Err() == nil { + t.Fatal("slow consumer/message limit did not end connection") + } + }) + } +} + +type realtimeCredential struct{ calls atomic.Int32 } + +func (c *realtimeCredential) Token(context.Context) (string, error) { + n := c.calls.Add(1) + if n == 1 { + return "sat-one", nil + } + return "sat-two", nil +} +func TestRealtimeAuthenticationAndRedirectSafety(t *testing.T) { + fixture := realtimeFixtures(t) + credential := &realtimeCredential{} + var auths []string + var mu sync.Mutex + server := realtimeServer(t, func(ws *websocket.Conn, r *http.Request) { + mu.Lock() + auths = append(auths, r.Header.Get("Authorization")) + mu.Unlock() + if strings.Contains(r.URL.String(), "sat-") { + t.Error("token leaked in URL") + } + writeRealtime(t, ws, fixture[1]) + _, _, _ = ws.Read(context.Background()) + }) + client := forward.NewClient(option.WithCredential(credential), option.WithBaseURL(server.URL+"/api/v1/forward")) + for i := 0; i < 2; i++ { + conn := realtimeConnect(t, client, forward.RealtimeConnectParams{}) + _ = conn.Close() + } + mu.Lock() + if !reflect.DeepEqual(auths, []string{"Bearer sat-one", "Bearer sat-two"}) { + t.Errorf("credential reused stale token: %v", auths) + } + mu.Unlock() + conn, err := client.Realtime.Connect(context.Background(), "conv_test", forward.RealtimeConnectParams{}, option.WithHeader("Authorization", "Bearer explicit")) + if err != nil { + t.Fatal(err) + } + _ = conn.Close() + if credential.calls.Load() != 2 { + t.Fatal("explicit header did not override credential") + } + targetCalls := atomic.Int32{} + target := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { targetCalls.Add(1); w.WriteHeader(401) })) + defer target.Close() + redirect := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + http.Redirect(w, r, target.URL, http.StatusTemporaryRedirect) + })) + defer redirect.Close() + client = realtimeClient(redirect.URL) + _, err = client.Realtime.Connect(context.Background(), "conv_test", forward.RealtimeConnectParams{}) + var api *forward.Error + if !errors.As(err, &api) || api.StatusCode != 307 || targetCalls.Load() != 0 { + t.Fatalf("redirect followed: %v", err) + } + for _, body := range []string{`{"code":"TOKEN_INVALID","message":"expired"}`, "proxy rejected"} { + bad := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("X-Request-Id", "handshake-failure") + w.WriteHeader(401) + _, _ = io.WriteString(w, body) + })) + client = realtimeClient(bad.URL) + _, err = client.Realtime.Connect(context.Background(), "conv_test", forward.RealtimeConnectParams{}) + bad.Close() + if !errors.As(err, &api) || api.StatusCode != 401 || api.RequestID != "handshake-failure" || api.Message == "" { + t.Fatalf("handshake error lost %v", err) + } + } +} + +type realtimeBlockedCredential struct{} + +func (realtimeBlockedCredential) Token(ctx context.Context) (string, error) { + <-ctx.Done() + return "", ctx.Err() +} + +func TestRealtimeHandshakeTimeoutIncludesCredential(t *testing.T) { + client := forward.NewClient(option.WithCredential(realtimeBlockedCredential{}), option.WithBaseURL("https://example.invalid/api/v1/forward")) + _, err := client.Realtime.Connect(context.Background(), "conv_test", forward.RealtimeConnectParams{HandshakeTimeout: 10 * time.Millisecond}) + if !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("credential wait escaped handshake timeout: %v", err) + } +} + +func TestRealtimeSharedServiceAccountCredentialTLSAndMiddleware(t *testing.T) { + fixture := realtimeFixtures(t) + exchanges := atomic.Int32{} + credential, err := convention.NewServiceAccountCredential(convention.ServiceAccountCredentialConfig{SAKey: "fake-sa-key", Region: "cn", Scopes: []string{"forward.access"}, HTTPClient: &http.Client{Transport: roundTripFunc(func(r *http.Request) (*http.Response, error) { + if r.URL.String() != "https://openapi.qoder.com.cn/api/v1/serviceToken/exchange" || r.Header.Get("Authorization") != "Bearer fake-sa-key" { + t.Error("unexpected SA exchange") + } + exchanges.Add(1) + return reply(r, 200, `{"access_token":"cached-sat","token_type":"Bearer","expires_in":3600}`), nil + })}}) + if err != nil { + t.Fatal(err) + } + server := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Header.Get("Authorization") != "Bearer cached-sat" || r.Header.Get("X-Middleware") != "present" { + t.Error("HTTP/WS authentication or middleware differed") + } + if strings.HasSuffix(r.URL.Path, "/models") { + w.Header().Set("Content-Type", "application/json") + _, _ = io.WriteString(w, `{"data":[]}`) + return + } + ws, err := websocket.Accept(w, r, nil) + if err != nil { + t.Error(err) + return + } + defer ws.CloseNow() + writeRealtime(t, ws, fixture[1]) + _, _, _ = ws.Read(context.Background()) + })) + defer server.Close() + client := forward.NewClient(option.WithCredential(credential), option.WithHTTPClient(server.Client()), option.WithBaseURL(server.URL+"/api/v1/forward"), option.WithMiddleware(func(r *http.Request, next option.MiddlewareNext) (*http.Response, error) { + r = r.Clone(r.Context()) + r.Header.Set("X-Middleware", "present") + return next(r) + })) + if _, err := client.Models.List(context.Background()); err != nil { + t.Fatal(err) + } + conn := realtimeConnect(t, client, forward.RealtimeConnectParams{}) + _ = conn.Close() + if exchanges.Load() != 1 { + t.Fatalf("REST/WS did not reuse shared SAT cache: exchanges=%d", exchanges.Load()) + } +} + +func TestRealtimeConcurrentWritesAndGracefulCapability(t *testing.T) { + fixture := realtimeFixtures(t) + received := make(chan int, 1) + server := realtimeServer(t, func(ws *websocket.Conn, _ *http.Request) { + ready := fixture[1] + ready["payload"].(map[string]any)["capabilities"] = map[string]any{"graceful_close": false} + writeRealtime(t, ws, ready) + seen := map[string]bool{} + for i := 0; i < 32; i++ { + event := readRealtime(t, ws) + if event == nil { + return + } + text := event["payload"].(map[string]any)["text"].(string) + seen[text] = true + } + received <- len(seen) + _, _, _ = ws.Read(context.Background()) + }) + conn := realtimeConnect(t, realtimeClient(server.URL), forward.RealtimeConnectParams{}) + var writers sync.WaitGroup + for i := 0; i < 32; i++ { + writers.Add(1) + go func(i int) { + defer writers.Done() + if err := conn.SendText(context.Background(), statusName(i)); err != nil { + t.Error(err) + } + }(i) + } + writers.Wait() + if count := <-received; count != 32 { + t.Fatalf("writes were lost/corrupted: %d", count) + } + if _, err := conn.CloseGracefully(context.Background()); err == nil || !strings.Contains(err.Error(), "does not support") { + t.Fatalf("unsupported graceful close was accepted: %v", err) + } +} + +func TestRealtimeLifetimeContextAndCloseCode(t *testing.T) { + for _, kind := range []string{"caller_cancel", "peer_close"} { + t.Run(kind, func(t *testing.T) { + fixture := realtimeFixtures(t) + release := make(chan struct{}) + server := realtimeServer(t, func(ws *websocket.Conn, _ *http.Request) { + writeRealtime(t, ws, fixture[1]) + <-release + if kind == "peer_close" { + _ = ws.Close(websocket.StatusServiceRestart, "restart") + } else { + _, _, _ = ws.Read(context.Background()) + } + }) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + client := realtimeClient(server.URL) + conn, err := client.Realtime.Connect(ctx, "conv_test", forward.RealtimeConnectParams{}) + if err != nil { + t.Fatal(err) + } + defer conn.Close() + _, _ = conn.Recv(context.Background()) + close(release) + if kind == "caller_cancel" { + cancel() + } + waitCtx, waitCancel := context.WithTimeout(context.Background(), time.Second) + defer waitCancel() + _, err = conn.Recv(waitCtx) + if kind == "caller_cancel" { + if !errors.Is(err, context.Canceled) { + t.Fatalf("lifetime context ignored: %v", err) + } + } else { + var transport *forward.RealtimeTransportError + if !errors.As(err, &transport) || transport.CloseCode != 1012 || transport.Reason != "restart" || transport.ConversationID != "conv_test" { + t.Fatalf("close diagnostics lost: %v", err) + } + } + }) + } +} diff --git a/forward/realtimeconversation.go b/forward/realtimeconversation.go new file mode 100644 index 0000000..5c8a431 --- /dev/null +++ b/forward/realtimeconversation.go @@ -0,0 +1,266 @@ +package forward + +import ( + "context" + "encoding/json" + "fmt" + "net/http" + "net/url" + "slices" + "strings" + "unicode" + "unicode/utf8" + + "github.com/QoderAI/qoder-cloud-agents-sdk-go/convention" + "github.com/QoderAI/qoder-cloud-agents-sdk-go/convention/apiquery" + "github.com/QoderAI/qoder-cloud-agents-sdk-go/convention/option" + "github.com/QoderAI/qoder-cloud-agents-sdk-go/convention/param" +) + +// RealtimeConversationService creates Realtime Conversations and queries their +// persisted history. Realtime is Beta; a Template must already have it enabled. +type RealtimeConversationService struct{ Options []option.RequestOption } + +func NewRealtimeConversationService(opts ...option.RequestOption) RealtimeConversationService { + return RealtimeConversationService{Options: slices.Clone(opts)} +} + +// New creates a Conversation. Supply a stable Idempotency-Key for this operation +// with IdempotencyKey or option.WithHeader; it is required by the service. +func (r *RealtimeConversationService) New(ctx context.Context, params RealtimeConversationNewParams, opts ...option.RequestOption) (res *RealtimeConversation, err error) { + if params.IdempotencyKey.Valid() { + opts = append([]option.RequestOption{option.WithHeader("Idempotency-Key", params.IdempotencyKey.Value)}, opts...) + } + cfg, err := convention.NewRequestConfig(ctx, http.MethodPost, "realtime/conversations", params, &res, slices.Concat(r.Options, opts)...) + if err != nil { + return nil, err + } + key := cfg.Request.Header.Get("Idempotency-Key") + if len(key) < 1 || len(key) > 256 || !utf8.ValidString(key) || strings.TrimSpace(key) != key || strings.ContainsFunc(key, unicode.IsControl) { + return nil, fmt.Errorf("Realtime creation requires Idempotency-Key of 1–256 UTF-8 bytes without surrounding whitespace or controls") + } + err = cfg.Execute() + return res, err +} + +type RealtimeConversationNewParams struct { + IdentityID string `json:"identity_id" api:"required"` + TemplateID string `json:"template_id" api:"required"` + Title param.Opt[string] `json:"title,omitzero"` + Metadata map[string]any `json:"metadata,omitzero"` + Config RealtimeConfigParam `json:"config,omitzero"` + IdempotencyKey param.Opt[string] `header:"Idempotency-Key,omitzero" json:"-"` + paramObj +} + +func (r RealtimeConversationNewParams) MarshalJSON() ([]byte, error) { + type shadow RealtimeConversationNewParams + return param.MarshalObject(r, (*shadow)(&r)) +} + +type RealtimeConfigParam struct { + Audio RealtimeAudioConfigParam `json:"audio,omitzero"` + paramObj +} + +func (r RealtimeConfigParam) MarshalJSON() ([]byte, error) { + type shadow RealtimeConfigParam + return param.MarshalObject(r, (*shadow)(&r)) +} + +type RealtimeAudioConfigParam struct { + Output RealtimeAudioOutputConfigParam `json:"output,omitzero"` + paramObj +} + +func (r RealtimeAudioConfigParam) MarshalJSON() ([]byte, error) { + type shadow RealtimeAudioConfigParam + return param.MarshalObject(r, (*shadow)(&r)) +} + +type RealtimeAudioOutputConfigParam struct { + Voice param.Opt[string] `json:"voice,omitzero"` + paramObj +} + +func (r RealtimeAudioOutputConfigParam) MarshalJSON() ([]byte, error) { + type shadow RealtimeAudioOutputConfigParam + return param.MarshalObject(r, (*shadow)(&r)) +} + +// RealtimeConversation timestamps remain text to preserve the wire precision. +type RealtimeConversation struct { + ID string `json:"id"` + Type string `json:"type"` + Status string `json:"status"` + Title *string `json:"title"` + Metadata map[string]any `json:"metadata"` + Config RealtimeConfig `json:"config"` + CreatedAt string `json:"created_at"` + UpdatedAt string `json:"updated_at"` +} +type RealtimeConfig struct { + Audio RealtimeAudioConfig `json:"audio"` +} +type RealtimeAudioConfig struct { + Output RealtimeAudioOutputConfig `json:"output"` +} +type RealtimeAudioOutputConfig struct { + Voice string `json:"voice"` +} + +type RealtimeConversationHistoryParams struct { + Limit param.Opt[int64] `query:"limit,omitzero" json:"-"` + Before param.Opt[string] `query:"before,omitzero" json:"-"` + // "message", "work", or "message,work"; the default includes both. + Types param.Opt[string] `query:"types,omitzero" json:"-"` + paramObj +} + +func (r RealtimeConversationHistoryParams) URLQuery() (url.Values, error) { + return apiquery.MarshalWithSettings(r, apiquery.QuerySettings{}) +} + +// History fetches a page of persisted events. Its cursor points towards older +// history, and is unrelated to WebSocket event IDs or sequence numbers. +func (r *RealtimeConversationService) History(ctx context.Context, conversationID string, params RealtimeConversationHistoryParams, opts ...option.RequestOption) (res *RealtimeHistoryPage, err error) { + if conversationID == "" { + return nil, fmt.Errorf("missing required conversation_id parameter") + } + if params.Limit.Valid() && (params.Limit.Value < 1 || params.Limit.Value > 100) { + return nil, fmt.Errorf("history limit must be between 1 and 100") + } + if params.Before.Valid() && params.Before.Value == "" { + return nil, fmt.Errorf("history before cursor must not be empty") + } + if params.Types.Valid() && params.Types.Value != "message" && params.Types.Value != "work" && params.Types.Value != "message,work" { + return nil, fmt.Errorf("history types must be message, work, or message,work") + } + err = convention.ExecuteNewRequest(ctx, http.MethodGet, "realtime/conversations/"+url.PathEscape(conversationID)+"/history", params, &res, slices.Concat(r.Options, opts)...) + if err == nil && res != nil { + if res.Page.HasMore && (res.Page.NextBefore == nil || *res.Page.NextBefore == "") { + return nil, &RealtimeProtocolError{Message: "history page has_more without next_before cursor"} + } + res.service = r + res.params = params + res.options = slices.Clone(opts) + res.conversationID = conversationID + } + return res, err +} + +type RealtimeHistoryConversation struct { + ID string `json:"id"` + Title *string `json:"title"` + InitializationStatus string `json:"initialization_status"` + Metadata map[string]any `json:"metadata"` + CreatedAt string `json:"created_at"` + UpdatedAt string `json:"updated_at"` +} + +// RealtimeHistoryEvent retains unknown fields through RawJSON. ID is a history +// ID (vevt_*), not an ephemeral WebSocket event_id. +type RealtimeHistoryEvent struct { + ID string `json:"id"` + Type string `json:"type"` + Status string `json:"status"` + OccurredAt string `json:"occurred_at"` + Role string `json:"role,omitempty"` + Text string `json:"text,omitempty"` + WorkID string `json:"work_id,omitempty"` + Objective string `json:"objective,omitempty"` + Result string `json:"result,omitempty"` + Error *RealtimeWorkError `json:"error,omitempty"` + TurnID string `json:"turn_id,omitempty"` + UserMessageEventID string `json:"user_message_event_id,omitempty"` + AnnouncementID string `json:"announcement_id,omitempty"` + MilestoneType string `json:"milestone_type,omitempty"` + raw string +} + +func (e *RealtimeHistoryEvent) UnmarshalJSON(b []byte) error { + type shadow RealtimeHistoryEvent + if err := json.Unmarshal(b, (*shadow)(e)); err != nil { + return err + } + e.raw = string(b) + return nil +} +func (e RealtimeHistoryEvent) RawJSON() string { return e.raw } + +type RealtimeHistoryPageInfo struct { + NextBefore *string `json:"next_before"` + HasMore bool `json:"has_more"` +} +type RealtimeHistoryPage struct { + Conversation RealtimeHistoryConversation `json:"conversation"` + Events []RealtimeHistoryEvent `json:"events"` + Page RealtimeHistoryPageInfo `json:"page"` + service *RealtimeConversationService + params RealtimeConversationHistoryParams + options []option.RequestOption + conversationID string +} + +func (p *RealtimeHistoryPage) HasNextPage() bool { + return p.Page.HasMore && p.Page.NextBefore != nil && *p.Page.NextBefore != "" +} + +// GetNextPage returns older history, or nil at the end. +func (p *RealtimeHistoryPage) GetNextPage(ctx context.Context) (*RealtimeHistoryPage, error) { + if !p.HasNextPage() { + return nil, nil + } + if p.service == nil { + return nil, fmt.Errorf("history page has no request configuration") + } + if p.params.Before.Valid() && p.params.Before.Value == *p.Page.NextBefore { + return nil, fmt.Errorf("history cursor did not advance") + } + params := p.params + params.Before = param.NewOpt(*p.Page.NextBefore) + return p.service.History(ctx, p.conversationID, params, p.options...) +} + +// HistoryAutoPaging iterates persisted history towards older records. +func (r *RealtimeConversationService) HistoryAutoPaging(ctx context.Context, conversationID string, params RealtimeConversationHistoryParams, opts ...option.RequestOption) *RealtimeHistoryAutoPager { + page, err := r.History(ctx, conversationID, params, opts...) + return &RealtimeHistoryAutoPager{ctx: ctx, page: page, err: err} +} + +type RealtimeHistoryAutoPager struct { + ctx context.Context + page *RealtimeHistoryPage + index int + current RealtimeHistoryEvent + err error + seen map[string]bool +} + +func (p *RealtimeHistoryAutoPager) Next() bool { + for p.err == nil && p.page != nil { + if p.index < len(p.page.Events) { + p.current = p.page.Events[p.index] + p.index++ + return true + } + if !p.page.HasNextPage() { + p.page = nil + return false + } + cursor := *p.page.Page.NextBefore + if p.seen == nil { + p.seen = map[string]bool{} + } + if p.seen[cursor] { + p.err = fmt.Errorf("history cursor cycle detected") + return false + } + p.seen[cursor] = true + p.page, p.err = p.page.GetNextPage(p.ctx) + p.index = 0 + } + return false +} +func (p *RealtimeHistoryAutoPager) Current() RealtimeHistoryEvent { return p.current } +func (p *RealtimeHistoryAutoPager) Err() error { return p.err } diff --git a/forward/realtimeconversation_test.go b/forward/realtimeconversation_test.go new file mode 100644 index 0000000..1666796 --- /dev/null +++ b/forward/realtimeconversation_test.go @@ -0,0 +1,123 @@ +package forward_test + +import ( + "context" + "encoding/json" + "errors" + "net/http" + "strings" + "testing" + + "github.com/QoderAI/qoder-cloud-agents-sdk-go/convention/option" + "github.com/QoderAI/qoder-cloud-agents-sdk-go/forward" +) + +func TestRealtimeConversationAPIContracts(t *testing.T) { + for _, status := range []int{200, 400} { + t.Run("create_"+statusName(status), func(t *testing.T) { + client := contractClient(t, "createRealtimeConversation", status) + res, err := client.Realtime.Conversations.New(context.Background(), contractParams[forward.RealtimeConversationNewParams](t, "createRealtimeConversation")) + checkError(t, status, err) + if status == 200 && (res.ID != "conv_contract" || res.CreatedAt != "2026-10-13T01:02:03.123456789Z" || res.Config.Audio.Output.Voice != "longanqian") { + t.Fatalf("conversation = %#v", res) + } + }) + t.Run("history_"+statusName(status), func(t *testing.T) { + client := contractClient(t, "getRealtimeConversationHistory", status) + res, err := client.Realtime.Conversations.History(context.Background(), pathSegment, contractParams[forward.RealtimeConversationHistoryParams](t, "getRealtimeConversationHistory")) + checkError(t, status, err) + if status == 200 && (res.Conversation.ID != "conv_contract" || len(res.Events) != 1 || res.Events[0].ID != "vevt_contract" || !strings.Contains(res.Events[0].RawJSON(), `"future": true`) || res.HasNextPage()) { + t.Fatalf("history = %#v", res) + } + }) + } +} + +func TestRealtimeCreationIdempotencyAndRetry(t *testing.T) { + for _, key := range []string{"", strings.Repeat("a", 257), " key", "key ", "key\n", strings.Repeat("你", 86)} { + t.Run("invalid_"+statusName(len(key)), func(t *testing.T) { + client := testClient(func(*http.Request) (*http.Response, error) { t.Fatal("invalid key reached transport"); return nil, nil }) + _, err := client.Realtime.Conversations.New(context.Background(), forward.RealtimeConversationNewParams{IdentityID: "idn", TemplateID: "tmpl", IdempotencyKey: forward.String(key)}) + if err == nil { + t.Fatal("expected key validation error") + } + }) + } + calls := 0 + client := testClient(func(r *http.Request) (*http.Response, error) { + calls++ + if r.Header.Get("Idempotency-Key") != "request-key" { + t.Fatal("retry changed operation key") + } + var body map[string]any + _ = json.NewDecoder(r.Body).Decode(&body) + if len(body) != 2 || body["identity_id"] != "idn" || body["template_id"] != "tmpl" { + t.Fatalf("unexpected body %#v", body) + } + if calls == 1 { + response := reply(r, 503, `{"error":{"code":"idempotency_unavailable"}}`) + response.Header.Set("Retry-After", "0.001") + return response, nil + } + return reply(r, 201, `{"id":"conv","type":"voice.conversation"}`), nil + }, option.WithMaxRetries(1)) + _, err := client.Realtime.Conversations.New(context.Background(), forward.RealtimeConversationNewParams{IdentityID: "idn", TemplateID: "tmpl", IdempotencyKey: forward.String("params-key")}, option.WithHeader("Idempotency-Key", "request-key")) + if err != nil || calls != 2 { + t.Fatalf("calls=%d err=%v", calls, err) + } + client = testClient(func(r *http.Request) (*http.Response, error) { + response := reply(r, 409, `{"error":{"code":"idempotency_key_in_progress","message":"busy"}}`) + response.Header.Set("Retry-After", "1") + return response, nil + }, option.WithMaxRetries(1)) + _, err = client.Realtime.Conversations.New(context.Background(), forward.RealtimeConversationNewParams{IdempotencyKey: forward.String("key")}) + var api *forward.Error + if !errors.As(err, &api) || api.Code != "idempotency_key_in_progress" || api.Response.Header.Get("Retry-After") != "1" { + t.Fatalf("lost conflict: %v", err) + } +} + +func TestRealtimeHistoryPaginationAndBounds(t *testing.T) { + calls := 0 + client := testClient(func(r *http.Request) (*http.Response, error) { + calls++ + if r.URL.Query().Get("types") != "work" || r.URL.Query().Get("limit") != "1" || r.Header.Get("X-Test") != "page-options" { + t.Fatal("pagination lost options") + } + switch calls { + case 1: + if r.URL.Query().Get("before") != "" { + t.Fatal("unexpected first cursor") + } + return reply(r, 200, `{"conversation":{"id":"conv"},"events":[{"id":"vevt_new","type":"voice.work.running"}],"page":{"has_more":true,"next_before":"older /?"}}`), nil + case 2: + if r.URL.Query().Get("before") != "older /?" { + t.Fatal("cursor was not encoded") + } + return reply(r, 200, `{"conversation":{"id":"conv"},"events":[],"page":{"has_more":true,"next_before":"oldest"}}`), nil + default: + if r.URL.Query().Get("before") != "oldest" { + t.Fatal("empty page ended pagination") + } + return reply(r, 200, `{"conversation":{"id":"conv"},"events":[{"id":"vevt_old","type":"voice.work.accepted"}],"page":{"has_more":false,"next_before":null}}`), nil + } + }) + pager := client.Realtime.Conversations.HistoryAutoPaging(context.Background(), "conv", forward.RealtimeConversationHistoryParams{Types: forward.String("work"), Limit: forward.Int(1)}, option.WithHeader("X-Test", "page-options")) + var ids []string + for pager.Next() { + ids = append(ids, pager.Current().ID) + } + if pager.Err() != nil || strings.Join(ids, ",") != "vevt_new,vevt_old" || calls != 3 { + t.Fatalf("ids=%v calls=%d err=%v", ids, calls, pager.Err()) + } + for _, params := range []forward.RealtimeConversationHistoryParams{{Limit: forward.Int(0)}, {Limit: forward.Int(101)}, {Before: forward.String("")}, {Types: forward.String("all")}} { + _, err := client.Realtime.Conversations.History(context.Background(), "conv", params) + if err == nil { + t.Fatal("invalid history filter accepted") + } + } + page := &forward.RealtimeHistoryPage{} + if next, err := page.GetNextPage(context.Background()); err != nil || next != nil { + t.Fatal("exhausted page should terminate") + } +} diff --git a/forward/realtimeevent.go b/forward/realtimeevent.go new file mode 100644 index 0000000..80c0ef8 --- /dev/null +++ b/forward/realtimeevent.go @@ -0,0 +1,238 @@ +package forward + +import ( + "encoding/base64" + "encoding/json" + "fmt" + "regexp" + "time" +) + +const RealtimeProtocolVersion = "voice.realtime.v1" + +var realtimeTimestampPattern = regexp.MustCompile(`^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d{1,9})?(?:Z|[+-](?:[01]\d|2[0-3]):[0-5]\d)$`) + +// RealtimeEvent is a server event. Payload and RawJSON preserve Beta extension +// fields; Timestamp preserves nanosecond precision. Sequence is not a cursor. +type RealtimeEvent struct { + Version string `json:"version"` + Type string `json:"type"` + EventID string `json:"event_id"` + Sequence int64 `json:"sequence"` + ConversationID string `json:"conversation_id"` + Timestamp string `json:"timestamp"` + WorkID string `json:"work_id,omitempty"` + AnnouncementID string `json:"announcement_id,omitempty"` + Payload json.RawMessage `json:"payload"` + raw string +} + +func (e RealtimeEvent) RawJSON() string { return e.raw } + +// DecodePayload decodes a known typed payload or an application's extension. +func (e RealtimeEvent) DecodePayload(dst any) error { return json.Unmarshal(e.Payload, dst) } + +// Audio decodes audio.delta PCM16. It does not decode WAV or play audio. +func (e RealtimeEvent) Audio() ([]byte, error) { + if e.Type != "audio.delta" { + return nil, fmt.Errorf("event is not audio.delta") + } + var p RealtimeAudioPayload + if err := e.DecodePayload(&p); err != nil { + return nil, err + } + pcm, err := base64.StdEncoding.DecodeString(p.Audio) + if err != nil || len(pcm) == 0 || len(pcm)%2 != 0 { + return nil, &RealtimeProtocolError{Message: "invalid PCM16 audio.delta"} + } + return pcm, nil +} + +type RealtimeAudioFormat struct { + Format string `json:"format"` + SampleRate int `json:"sample_rate"` + Channels int `json:"channels"` +} +type RealtimeCapabilities struct { + GracefulClose bool `json:"graceful_close"` +} +type RealtimeReadyPayload struct { + Config RealtimeConfig `json:"config"` + Provider string `json:"provider"` + InputAudio RealtimeAudioFormat `json:"input_audio"` + OutputAudio RealtimeAudioFormat `json:"output_audio"` + Capabilities RealtimeCapabilities `json:"capabilities"` +} +type RealtimeStatePayload struct { + State string `json:"state"` +} +type RealtimeReasonPayload struct { + Reason string `json:"reason"` +} +type RealtimeTranscriptPayload struct { + Role string `json:"role"` + Text string `json:"text"` + ItemID string `json:"item_id,omitempty"` + ResponseID string `json:"response_id,omitempty"` +} +type RealtimeAudioPayload struct { + Audio string `json:"audio"` +} +type RealtimeWorkAcceptedPayload struct { + Status string `json:"status"` + Objective string `json:"objective"` +} +type RealtimeWorkRunningPayload struct { + Status string `json:"status"` +} +type RealtimeWorkProgressPayload struct { + Kind string `json:"kind"` + Detail string `json:"detail"` + Title string `json:"title,omitempty"` + IsError bool `json:"is_error,omitempty"` + EventID string `json:"event_id,omitempty"` +} +type RealtimeWorkMilestonePayload struct { + AnnouncementID string `json:"announcement_id"` + MilestoneType string `json:"milestone_type"` + Summary string `json:"summary"` +} +type RealtimeWorkCompletedPayload struct { + Status string `json:"status"` + Result string `json:"result"` + EventID string `json:"event_id,omitempty"` +} +type RealtimeWorkError struct { + Code string `json:"code"` +} +type RealtimeWorkFailedPayload struct { + Status string `json:"status"` + Error RealtimeWorkError `json:"error"` + EventID string `json:"event_id,omitempty"` +} +type RealtimeClosedPayload struct { + RequestID string `json:"request_id"` + Outcome string `json:"outcome"` +} +type RealtimeErrorPayload struct { + Code string `json:"code"` + Message string `json:"message"` + Retryable *bool `json:"retryable,omitempty"` + RetryAfterMS *int64 `json:"retry_after_ms,omitempty"` +} + +// RealtimeClientEvent is always encoded as a JSON text frame. Use SendEvent for +// protocol extensions; convenience methods provide validated public events. +type RealtimeClientEvent struct { + Version string `json:"version"` + Type string `json:"type"` + Payload map[string]any `json:"payload"` +} +type RealtimePlaybackReceipt struct { + WorkID string `json:"work_id,omitempty"` + AnnouncementID string `json:"announcement_id,omitempty"` +} + +type RealtimeProtocolError struct{ Message string } + +func (e *RealtimeProtocolError) Error() string { return "qca realtime protocol: " + e.Message } + +type RealtimeServerError struct { + RealtimeErrorPayload + ConversationID string +} + +func (e *RealtimeServerError) Error() string { + return fmt.Sprintf("qca realtime: %s: %s (conversation_id=%s)", e.Code, e.Message, e.ConversationID) +} + +func decodeRealtimeEvent(b []byte, conversationID string) (RealtimeEvent, error) { + var e RealtimeEvent + var fields map[string]json.RawMessage + if err := json.Unmarshal(b, &fields); err != nil { + return e, &RealtimeProtocolError{Message: "invalid JSON event"} + } + if fields == nil { + return e, &RealtimeProtocolError{Message: "event must be an object"} + } + if err := json.Unmarshal(b, &e); err != nil { + return e, &RealtimeProtocolError{Message: "invalid event envelope"} + } + if e.Version != RealtimeProtocolVersion || e.Type == "" || e.EventID == "" || e.ConversationID != conversationID || fields["sequence"] == nil || string(fields["sequence"]) == "null" || e.Sequence < 0 { + return e, &RealtimeProtocolError{Message: "missing or invalid event envelope fields"} + } + if _, err := time.Parse(time.RFC3339Nano, e.Timestamp); err != nil || !realtimeTimestampPattern.MatchString(e.Timestamp) { + return e, &RealtimeProtocolError{Message: "invalid event timestamp"} + } + var payload map[string]json.RawMessage + if err := json.Unmarshal(e.Payload, &payload); err != nil || payload == nil { + return e, &RealtimeProtocolError{Message: "event payload must be an object"} + } + required := map[string][]string{ + "voice.state": {"state"}, "voice.ready": {"config", "provider", "input_audio", "output_audio", "capabilities"}, "voice.replaced": {"reason"}, + "transcript.delta": {"role", "text"}, "transcript.final": {"role", "text"}, "audio.delta": {"audio"}, "playback.interrupt": {"reason"}, + "work.accepted": {"status", "objective"}, "work.running": {"status"}, "work.progress": {"kind", "detail"}, "work.milestone": {"announcement_id", "milestone_type", "summary"}, + "work.completed": {"status", "result"}, "work.failed": {"status", "error"}, "connection.closed": {"request_id", "outcome"}, "error": {"code", "message"}, + } + for _, key := range required[e.Type] { + if payload[key] == nil || string(payload[key]) == "null" { + return e, &RealtimeProtocolError{Message: "missing " + e.Type + " payload field " + key} + } + } + var typed any + switch e.Type { + case "voice.ready": + typed = &RealtimeReadyPayload{} + case "voice.state": + typed = &RealtimeStatePayload{} + case "voice.replaced", "playback.interrupt": + typed = &RealtimeReasonPayload{} + case "transcript.delta", "transcript.final": + typed = &RealtimeTranscriptPayload{} + case "audio.delta": + typed = &RealtimeAudioPayload{} + case "work.accepted": + typed = &RealtimeWorkAcceptedPayload{} + case "work.running": + typed = &RealtimeWorkRunningPayload{} + case "work.progress": + typed = &RealtimeWorkProgressPayload{} + case "work.milestone": + typed = &RealtimeWorkMilestonePayload{} + case "work.completed": + typed = &RealtimeWorkCompletedPayload{} + case "work.failed": + typed = &RealtimeWorkFailedPayload{} + case "connection.closed": + typed = &RealtimeClosedPayload{} + case "error": + typed = &RealtimeErrorPayload{} + } + if typed != nil { + if err := json.Unmarshal(e.Payload, typed); err != nil { + return e, &RealtimeProtocolError{Message: "invalid " + e.Type + " payload types"} + } + } + if e.Type == "work.failed" { + var failure RealtimeWorkFailedPayload + _ = json.Unmarshal(e.Payload, &failure) + if failure.Error.Code == "" { + return e, &RealtimeProtocolError{Message: "missing work.failed error code"} + } + } + if expected, ok := map[string]string{"work.accepted": "accepted", "work.running": "running", "work.completed": "completed", "work.failed": "failed"}[e.Type]; ok { + var status string + if json.Unmarshal(payload["status"], &status) != nil || status != expected { + return e, &RealtimeProtocolError{Message: "invalid " + e.Type + " status"} + } + } + if e.Type == "error" { + var failure RealtimeErrorPayload + _ = json.Unmarshal(e.Payload, &failure) + if failure.Code == "" || (failure.RetryAfterMS != nil && *failure.RetryAfterMS < 0) { + return e, &RealtimeProtocolError{Message: "invalid error code or retry_after_ms"} + } + } + e.raw = string(b) + return e, nil +} diff --git a/forward/test_helpers_test.go b/forward/test_helpers_test.go index 21359e2..f4835e4 100644 --- a/forward/test_helpers_test.go +++ b/forward/test_helpers_test.go @@ -86,8 +86,8 @@ func operations(t *testing.T) []operationCase { Cases []operationCase `json:"cases"` } readJSON(t, "testdata/api-operation-cases.json", &fixture) - if fixture.Count != 116 || len(fixture.Cases) != 116 { - t.Fatal("Forward API scope must contain 116 operations") + if fixture.Count != 118 || len(fixture.Cases) != 118 { + t.Fatal("Forward API scope must contain 118 operations") } return fixture.Cases } diff --git a/forward/testdata/api-contracts.json b/forward/testdata/api-contracts.json index 9bf5338..03196d7 100644 --- a/forward/testdata/api-contracts.json +++ b/forward/testdata/api-contracts.json @@ -3251,6 +3251,69 @@ "raw": false, "empty": false }, + { + "operation_id": "createRealtimeConversation", + "service": "RealtimeConversationService", + "entry": "Realtime.Conversations", + "name": "New", + "response": { + "id": "conv_contract", + "type": "voice.conversation", + "status": "ready", + "title": "test", + "metadata": { + "source": "sdk" + }, + "config": { + "audio": { + "output": { + "voice": "longanqian" + } + } + }, + "created_at": "2026-10-13T01:02:03.123456789Z", + "updated_at": "2026-10-13T01:02:03.123456789Z" + }, + "stream": false, + "download": false, + "raw": false, + "empty": false + }, + { + "operation_id": "getRealtimeConversationHistory", + "service": "RealtimeConversationService", + "entry": "Realtime.Conversations", + "name": "History", + "response": { + "conversation": { + "id": "conv_contract", + "title": "test", + "initialization_status": "ready", + "metadata": {}, + "created_at": "2026-10-13T01:02:03Z", + "updated_at": "2026-10-13T01:02:03Z" + }, + "events": [ + { + "id": "vevt_contract", + "type": "voice.user_message.completed", + "status": "completed", + "text": "hello", + "occurred_at": "2026-10-13T01:02:03Z", + "role": "user", + "future": true + } + ], + "page": { + "has_more": false, + "next_before": null + } + }, + "stream": false, + "download": false, + "raw": false, + "empty": false + }, { "operation_id": "createServiceAccountToken", "entry": "ServiceAccountTokens", diff --git a/forward/testdata/api-operation-cases.json b/forward/testdata/api-operation-cases.json index 5be5847..dc72c97 100644 --- a/forward/testdata/api-operation-cases.json +++ b/forward/testdata/api-operation-cases.json @@ -1,6 +1,6 @@ { "schema_version": 1, - "operation_count": 116, + "operation_count": 118, "cases": [ { "operation_id": "addSessionResource", @@ -4925,6 +4925,73 @@ "request_body": null, "source": "cloud-agent/cloud-agents-api-doc origin/master 1b3f3d023b8631395c3f65577a507050d3893d12 (2026-10-09)" }, + { + "operation_id": "createRealtimeConversation", + "resource": "Realtime.Conversations", + "http_method": "POST", + "path": "/realtime/conversations", + "parameters": [ + { + "wire_name": "Idempotency-Key", + "location": "header", + "required": true, + "value": "realtime-contract" + } + ], + "request_body": { + "kind": "json", + "value": { + "identity_id": "idn_contract", + "template_id": "tmpl_contract", + "metadata": { + "source": "sdk" + }, + "config": { + "audio": { + "output": { + "voice": "longanqian" + } + } + }, + "title": "test" + } + }, + "source": "knowledge/api_specs/realtime/Cloud Agents Realtime API.md" + }, + { + "operation_id": "getRealtimeConversationHistory", + "resource": "Realtime.Conversations", + "http_method": "GET", + "path": "/realtime/conversations/{conversation_id}/history", + "parameters": [ + { + "wire_name": "conversation_id", + "location": "path", + "required": false, + "value": "conv_contract" + }, + { + "wire_name": "limit", + "location": "query", + "required": false, + "value": 50 + }, + { + "wire_name": "before", + "location": "query", + "required": false, + "value": "cursor_contract" + }, + { + "wire_name": "types", + "location": "query", + "required": false, + "value": "message,work" + } + ], + "request_body": null, + "source": "knowledge/api_specs/realtime/Cloud Agents Realtime API.md" + }, { "operation_id": "createServiceAccountToken", "http_method": "POST", diff --git a/forward/testdata/realtime-events.json b/forward/testdata/realtime-events.json new file mode 100644 index 0000000..fd39eef --- /dev/null +++ b/forward/testdata/realtime-events.json @@ -0,0 +1,277 @@ +{ + "source": "knowledge/api_specs/realtime/Cloud Agents Realtime API.md; voice gateway 8f50fa0494ab6f57bad1b3b839aef8a645ef6810", + "events": [ + { + "version": "voice.realtime.v1", + "type": "voice.state", + "event_id": "evt_1", + "sequence": 1, + "conversation_id": "conv_test", + "timestamp": "2026-10-13T01:02:03.123456789Z", + "payload": { + "state": "connecting" + }, + "future_envelope": "preserved" + }, + { + "version": "voice.realtime.v1", + "type": "voice.ready", + "event_id": "evt_2", + "sequence": 2, + "conversation_id": "conv_test", + "timestamp": "2026-10-13T01:02:03.123456789Z", + "payload": { + "config": { + "audio": { + "output": { + "voice": "longanqian" + } + } + }, + "provider": "qoder", + "input_audio": { + "format": "pcm16", + "sample_rate": 16000, + "channels": 1 + }, + "output_audio": { + "format": "pcm16", + "sample_rate": 24000, + "channels": 1 + }, + "capabilities": { + "graceful_close": true + } + }, + "future_envelope": "preserved" + }, + { + "version": "voice.realtime.v1", + "type": "transcript.delta", + "event_id": "evt_3", + "sequence": 3, + "conversation_id": "conv_test", + "timestamp": "2026-10-13T01:02:03.123456789Z", + "payload": { + "role": "assistant", + "text": "\u4f60", + "item_id": "item_one", + "response_id": "resp_one" + }, + "future_envelope": "preserved" + }, + { + "version": "voice.realtime.v1", + "type": "transcript.final", + "event_id": "evt_4", + "sequence": 4, + "conversation_id": "conv_test", + "timestamp": "2026-10-13T01:02:03.123456789Z", + "payload": { + "role": "assistant", + "text": "\u4f60\u597d" + }, + "future_envelope": "preserved" + }, + { + "version": "voice.realtime.v1", + "type": "audio.delta", + "event_id": "evt_5", + "sequence": 5, + "conversation_id": "conv_test", + "timestamp": "2026-10-13T01:02:03.123456789Z", + "payload": { + "audio": "AAABAA==" + }, + "work_id": "work_one", + "announcement_id": "ann_one", + "future_envelope": "preserved" + }, + { + "version": "voice.realtime.v1", + "type": "audio.done", + "event_id": "evt_6", + "sequence": 6, + "conversation_id": "conv_test", + "timestamp": "2026-10-13T01:02:03.123456789Z", + "payload": {}, + "work_id": "work_one", + "announcement_id": "ann_one", + "future_envelope": "preserved" + }, + { + "version": "voice.realtime.v1", + "type": "playback.interrupt", + "event_id": "evt_7", + "sequence": 7, + "conversation_id": "conv_test", + "timestamp": "2026-10-13T01:02:03.123456789Z", + "payload": { + "reason": "barge_in" + }, + "work_id": "work_one", + "announcement_id": "ann_one", + "future_envelope": "preserved" + }, + { + "version": "voice.realtime.v1", + "type": "work.accepted", + "event_id": "evt_8", + "sequence": 8, + "conversation_id": "conv_test", + "timestamp": "2026-10-13T01:02:03.123456789Z", + "payload": { + "status": "accepted", + "objective": "check" + }, + "work_id": "work_one", + "announcement_id": "ann_one", + "future_envelope": "preserved" + }, + { + "version": "voice.realtime.v1", + "type": "work.running", + "event_id": "evt_9", + "sequence": 9, + "conversation_id": "conv_test", + "timestamp": "2026-10-13T01:02:03.123456789Z", + "payload": { + "status": "running" + }, + "work_id": "work_one", + "announcement_id": "ann_one", + "future_envelope": "preserved" + }, + { + "version": "voice.realtime.v1", + "type": "work.progress", + "event_id": "evt_10", + "sequence": 10, + "conversation_id": "conv_test", + "timestamp": "2026-10-13T01:02:03.123456789Z", + "payload": { + "kind": "tool", + "detail": "read file", + "title": "Read", + "is_error": false, + "event_id": "evt_forward" + }, + "work_id": "work_one", + "announcement_id": "ann_one", + "future_envelope": "preserved" + }, + { + "version": "voice.realtime.v1", + "type": "work.milestone", + "event_id": "evt_11", + "sequence": 11, + "conversation_id": "conv_test", + "timestamp": "2026-10-13T01:02:03.123456789Z", + "payload": { + "announcement_id": "ann_one", + "milestone_type": "progress", + "summary": "done" + }, + "work_id": "work_one", + "announcement_id": "ann_one", + "future_envelope": "preserved" + }, + { + "version": "voice.realtime.v1", + "type": "work.completed", + "event_id": "evt_12", + "sequence": 12, + "conversation_id": "conv_test", + "timestamp": "2026-10-13T01:02:03.123456789Z", + "payload": { + "status": "completed", + "result": "ok", + "event_id": "evt_forward" + }, + "work_id": "work_one", + "announcement_id": "ann_one", + "future_envelope": "preserved" + }, + { + "version": "voice.realtime.v1", + "type": "work.failed", + "event_id": "evt_13", + "sequence": 13, + "conversation_id": "conv_test", + "timestamp": "2026-10-13T01:02:03.123456789Z", + "payload": { + "status": "failed", + "error": { + "code": "task_error" + } + }, + "work_id": "work_one", + "announcement_id": "ann_one", + "future_envelope": "preserved" + }, + { + "version": "voice.realtime.v1", + "type": "pong", + "event_id": "evt_14", + "sequence": 14, + "conversation_id": "conv_test", + "timestamp": "2026-10-13T01:02:03.123456789Z", + "payload": {}, + "future_envelope": "preserved" + }, + { + "version": "voice.realtime.v1", + "type": "error", + "event_id": "evt_15", + "sequence": 15, + "conversation_id": "conv_test", + "timestamp": "2026-10-13T01:02:03.123456789Z", + "payload": { + "code": "work_busy", + "message": "busy", + "retryable": true, + "retry_after_ms": 20 + }, + "future_envelope": "preserved" + }, + { + "version": "voice.realtime.v1", + "type": "future.extension", + "event_id": "evt_16", + "sequence": 16, + "conversation_id": "conv_test", + "timestamp": "2026-10-13T01:02:03.123456789Z", + "payload": { + "future": { + "value": true + } + }, + "future_envelope": "preserved" + }, + { + "version": "voice.realtime.v1", + "type": "connection.closed", + "event_id": "evt_17", + "sequence": 17, + "conversation_id": "conv_test", + "timestamp": "2026-10-13T01:02:03.123456789Z", + "payload": { + "request_id": "fixture_close", + "outcome": "saved" + }, + "future_envelope": "preserved" + }, + { + "version": "voice.realtime.v1", + "type": "voice.replaced", + "event_id": "evt_18", + "sequence": 18, + "conversation_id": "conv_test", + "timestamp": "2026-10-13T01:02:03.123456789Z", + "payload": { + "reason": "new_connection" + }, + "future_envelope": "preserved" + } + ] +} diff --git a/go.mod b/go.mod index e98dd8f..648e6ce 100644 --- a/go.mod +++ b/go.mod @@ -2,4 +2,7 @@ module github.com/QoderAI/qoder-cloud-agents-sdk-go go 1.23 -require gopkg.in/validator.v2 v2.0.1 +require ( + github.com/coder/websocket v1.8.14 + gopkg.in/validator.v2 v2.0.1 +) diff --git a/go.sum b/go.sum index d1929e4..59d1e3b 100644 --- a/go.sum +++ b/go.sum @@ -1,3 +1,5 @@ +github.com/coder/websocket v1.8.14 h1:9L0p0iKiNOibykf283eHkKUHHrpG7f65OE3BhhO7v9g= +github.com/coder/websocket v1.8.14/go.mod h1:NX3SzP+inril6yawo5CQXx8+fk145lPDC6pumgx0mVg= github.com/kr/pretty v0.2.1 h1:Fmg33tUaq4/8ym9TJN1x7sLJnHVwhP33CNkpYV/7rwI= github.com/kr/pretty v0.2.1/go.mod h1:ipq/a2n7PKx3OHsz4KJII5eveXtPO4qwEXGdVfWzfnI= github.com/kr/text v0.1.0 h1:45sCR5RtlFHMR4UwH9sdQ5TC8v0qDQCHnXt+kaKSTVE= diff --git a/integration/realtime_live_test.go b/integration/realtime_live_test.go new file mode 100644 index 0000000..3dc4563 --- /dev/null +++ b/integration/realtime_live_test.go @@ -0,0 +1,343 @@ +//go:build live + +package integration_test + +import ( + "context" + "errors" + "fmt" + "os" + "path/filepath" + "strings" + "testing" + "time" + + "github.com/QoderAI/qoder-cloud-agents-sdk-go/convention" + "github.com/QoderAI/qoder-cloud-agents-sdk-go/convention/option" + "github.com/QoderAI/qoder-cloud-agents-sdk-go/forward" + "github.com/QoderAI/qoder-cloud-agents-sdk-go/internal/testsupport" +) + +func realtimeEnvironment(path string, environment map[string]string) (map[string]string, error) { + values, err := liveEnvironment(path, environment) + if err != nil { + return nil, err + } + for _, name := range []string{"QODER_REALTIME_LIVE_ALLOW_WRITE", "QODER_REALTIME_LIVE_ALLOW_EXECUTION"} { + if values[name] != "true" { + return nil, fmt.Errorf("%s=true is required", name) + } + } + for _, region := range []string{"CN", "GLOBAL"} { + sat, key := "QODER_REALTIME_"+region+"_SAT", "QODER_SA_"+region+"_KEY" + if strings.TrimSpace(values[sat]) == "" && strings.TrimSpace(values[key]) == "" { + return nil, fmt.Errorf("%s or %s is required", sat, key) + } + pat := "QODER_FORWARD_" + region + "_PAT" + if strings.TrimSpace(values[pat]) == "" { + return nil, fmt.Errorf("%s is required", pat) + } + for _, mode := range []string{"PAT", "SA"} { + name := "QODER_REALTIME_" + region + "_" + mode + "_TEMPLATE_ID" + if strings.TrimSpace(values[name]) == "" { + return nil, fmt.Errorf("%s is required", name) + } + } + } + return values, nil +} + +func TestRealtimeConfiguration(t *testing.T) { + path := filepath.Join(t.TempDir(), "no-env-file") + valid := map[string]string{"QODER_SA_CN_KEY": "fake-cn-key", "QODER_SA_GLOBAL_KEY": "fake-global-key", "QODER_FORWARD_CN_PAT": "fake-cn-pat", "QODER_FORWARD_GLOBAL_PAT": "fake-global-pat", "QODER_REALTIME_LIVE_ALLOW_WRITE": "true", "QODER_REALTIME_LIVE_ALLOW_EXECUTION": "true"} + for _, region := range []string{"CN", "GLOBAL"} { + for _, mode := range []string{"PAT", "SA"} { + valid["QODER_REALTIME_"+region+"_"+mode+"_TEMPLATE_ID"] = "tmpl_fake" + } + } + if _, err := realtimeEnvironment(path, valid); err != nil { + t.Fatal(err) + } + for key := range valid { + copy := map[string]string{} + for k, v := range valid { + copy[k] = v + } + copy[key] = "" + if _, err := realtimeEnvironment(path, copy); err == nil { + t.Fatalf("missing %s did not fail before requests", key) + } + } + withSAT := map[string]string{} + for key, value := range valid { + withSAT[key] = value + } + for _, region := range []string{"CN", "GLOBAL"} { + withSAT["QODER_SA_"+region+"_KEY"] = "" + withSAT["QODER_REALTIME_"+region+"_SAT"] = "fake-bound-sat" + } + if _, err := realtimeEnvironment(path, withSAT); err != nil { + t.Fatal("a supplied bound SAT must not require a platform SA Key") + } + pcmPath := filepath.Join(t.TempDir(), "input.pcm") + for _, content := range [][]byte{nil, {0}, {0, 0, 1, 0}} { + if err := os.WriteFile(pcmPath, content, 0600); err != nil { + t.Fatal(err) + } + pcm, err := realtimePCMInput(pcmPath) + if len(content) == 4 { + if err != nil || len(pcm) != 4 { + t.Fatal("valid raw PCM fixture rejected") + } + } else if err == nil { + t.Fatal("invalid raw PCM fixture accepted") + } + } +} + +func realtimePCMInput(path string) ([]byte, error) { + if path == "" { + return nil, nil + } + pcm, err := os.ReadFile(path) + if err != nil { + return nil, fmt.Errorf("cannot read Realtime PCM fixture") + } + if len(pcm) == 0 || len(pcm)%2 != 0 || len(pcm) > 30*16000*2 { + return nil, fmt.Errorf("Realtime PCM fixture must be nonempty complete PCM16 samples and at most 30 seconds") + } + if len(pcm) >= 12 && string(pcm[:4]) == "RIFF" && string(pcm[8:12]) == "WAVE" { + return nil, fmt.Errorf("Realtime PCM fixture must be raw PCM16, not a WAV file") + } + return pcm, nil +} + +// TestRealtimeLive is explicit and never runs in the offline suite. Both regions +// and both auth modes are preflighted before any test Identity is created. +func TestRealtimeLive(t *testing.T) { + path := os.Getenv("LIVE_ENV_FILE") + if path == "" { + path = filepath.Join("..", ".env.live") + } + environment := map[string]string{} + for _, entry := range os.Environ() { + name, value, _ := strings.Cut(entry, "=") + environment[name] = value + } + values, err := realtimeEnvironment(path, environment) + if err != nil { + t.Fatal(err) + } + pcmPath := values["QODER_REALTIME_LIVE_PCM_FILE"] + if pcmPath == "" { + pcmPath = values["QODER_REALTIME_PCM_FILE"] + } + pcm, err := realtimePCMInput(pcmPath) + if err != nil { + t.Fatal(err) + } + for _, region := range []string{"cn", "global"} { + for _, mode := range []string{"pat", "sa"} { + t.Run(region+"/"+mode, func(t *testing.T) { + host := "https://api.qoder.com" + if region == "cn" { + host = "https://api.qoder.com.cn" + } + opts := []option.RequestOption{option.WithBaseURL(host + "/api/v1/forward"), option.WithMaxRetries(0), option.WithRequestTimeout(30 * time.Second), testsupport.RequestLog(t)} + if mode == "pat" { + opts = append(opts, option.WithPAT(values["QODER_FORWARD_"+strings.ToUpper(region)+"_PAT"])) + } else if sat := values["QODER_REALTIME_"+strings.ToUpper(region)+"_SAT"]; strings.TrimSpace(sat) != "" { + opts = append(opts, option.WithAuthMode("sa"), option.WithAuthToken(sat)) + t.Log("auth_mode=sa source=forward_bound_sat") + } else { + credential, err := convention.NewServiceAccountCredential(convention.ServiceAccountCredentialConfig{SAKey: values["QODER_SA_"+strings.ToUpper(region)+"_KEY"], Region: region, Scopes: []string{"forward.access"}}) + saCheck(t, err) + opts = append(opts, option.WithAuthMode("sa"), option.WithCredential(credential)) + t.Log("auth_mode=sa source=platform_credential") + } + client := forward.NewClient(opts...) + ctx, cancel := context.WithTimeout(context.Background(), 4*time.Minute) + defer cancel() + templateID := values["QODER_REALTIME_"+strings.ToUpper(region)+"_"+strings.ToUpper(mode)+"_TEMPLATE_ID"] + identity, err := client.Identities.New(ctx, forward.IdentityNewParams{ExternalID: "sdk-go-realtime-" + testsupport.Marker(t), IdempotencyKey: forward.String("sdk-go-realtime-identity-" + testsupport.Marker(t))}) + saCheck(t, err) + t.Logf("identity_id=%s template_id=%s", identity.ID, templateID) + saCleanup(t, "realtime identity "+identity.ID, func(cleanCtx context.Context) error { + // The Conversation's primary Forward Session belongs to this unique + // test Identity. Finish/archive it before deleting the Identity. + pager := client.Sessions.ListAutoPaging(cleanCtx, forward.SessionListParams{IdentityIDs: []string{identity.ID}, Limit: forward.Int(100)}) + var cleanupErrors []error + for pager.Next() { + session := pager.Current() + if err := saFinishSession(cleanCtx, func() (string, error) { + s, err := client.Sessions.Get(cleanCtx, session.ID) + if err != nil { + return "", err + } + return string(s.Status), nil + }, func() error { + _, err := client.Sessions.Cancel(cleanCtx, session.ID, forward.SessionCancelParams{}) + return err + }, func() error { + _, err := client.Sessions.Archive(cleanCtx, session.ID, forward.SessionArchiveParams{}) + return err + }); err != nil { + cleanupErrors = append(cleanupErrors, err) + } + } + if pager.Err() != nil { + cleanupErrors = append(cleanupErrors, pager.Err()) + } + _, err := client.Identities.Delete(cleanCtx, identity.ID) + if err != nil { + cleanupErrors = append(cleanupErrors, err) + } + return errors.Join(cleanupErrors...) + }) + conversation, err := client.Realtime.Conversations.New(ctx, forward.RealtimeConversationNewParams{IdentityID: identity.ID, TemplateID: templateID, IdempotencyKey: forward.String("sdk-go-realtime-conversation-" + testsupport.Marker(t)), Title: forward.String("SDK Realtime acceptance"), Metadata: map[string]any{"source": "sdk-go-live"}}) + saCheck(t, err) + if conversation.ID == "" || conversation.Type != "voice.conversation" || conversation.Status != "ready" { + t.Fatal("invalid Conversation response") + } + t.Logf("conversation_id=%s retained=true", conversation.ID) + conn, err := client.Realtime.Connect(ctx, conversation.ID, forward.RealtimeConnectParams{}) + saCheck(t, err) + defer conn.Close() + t.Logf("websocket conversation_id=%s request_id=%s", conn.ConversationID, conn.RequestID) + saCheck(t, conn.Ping(ctx)) + seenReady, seenPong := false, false + for !seenPong { + event, err := conn.Recv(ctx) + saCheck(t, err) + if event.Type == "voice.ready" { + seenReady = true + } + if event.Type == "pong" { + seenPong = true + } + realtimeLiveError(t, event) + } + if !seenReady { + t.Fatal("voice.ready was consumed during Connect") + } + saCheck(t, conn.SendText(ctx, "请用一句话简单打招呼。")) + audioBytes := realtimeLiveTurn(t, ctx, conn, false) + if len(pcm) > 0 { + sent := make(chan error, 1) + go func() { + for start := 0; start < len(pcm); start += 640 { + if err := conn.AppendAudio(ctx, pcm[start:min(start+640, len(pcm))]); err != nil { + sent <- err + return + } + timer := time.NewTimer(20 * time.Millisecond) + select { + case <-ctx.Done(): + timer.Stop() + sent <- ctx.Err() + return + case <-timer.C: + } + } + sent <- nil + }() + audioBytes += realtimeLiveTurn(t, ctx, conn, true) + select { + case err := <-sent: + saCheck(t, err) + case <-ctx.Done(): + saCheck(t, ctx.Err()) + } + t.Logf("conversation_id=%s input_pcm_bytes=%d user_transcript_final=true", conversation.ID, len(pcm)) + } + result, err := conn.CloseGracefully(ctx) + saCheck(t, err) + // response.done can already have committed the message before close, + // legitimately yielding no_active_response/no_content. The SDK validates + // the acknowledgement; persisted user+assistant history remains mandatory. + t.Logf("conversation_id=%s transcript_final=true audio_done=true audio_bytes=%d close_outcome=%s", conversation.ID, audioBytes, result.Outcome) + for _, types := range []string{"message", "work", "message,work"} { + pager := client.Realtime.Conversations.HistoryAutoPaging(ctx, conversation.ID, forward.RealtimeConversationHistoryParams{Limit: forward.Int(1), Types: forward.String(types)}) + ids := map[string]bool{} + count := 0 + seenUser, seenAssistant := false, false + for pager.Next() { + event := pager.Current() + if event.ID == "" || ids[event.ID] || event.OccurredAt == "" { + t.Fatal("invalid/duplicate persisted history ID or timestamp") + } + ids[event.ID] = true + count++ + seenUser = seenUser || strings.HasPrefix(event.Type, "voice.user_message.") + seenAssistant = seenAssistant || strings.HasPrefix(event.Type, "voice.assistant_message.") + if types == "message" && !strings.HasPrefix(event.Type, "voice.user_message.") && !strings.HasPrefix(event.Type, "voice.assistant_message.") { + t.Fatal("message filter returned a Work record") + } + if types == "work" && !strings.HasPrefix(event.Type, "voice.work.") { + t.Fatal("work filter returned a message record") + } + } + saCheck(t, pager.Err()) + if types != "work" && (count < 2 || !seenUser || !seenAssistant) { + t.Fatal("history did not contain persisted user/assistant messages with limit=1 pagination") + } + t.Logf("conversation_id=%s history_filter=%s count=%d", conversation.ID, types, count) + } + }) + } + } +} + +func realtimeLiveTurn(t *testing.T, ctx context.Context, conn *forward.RealtimeConnection, requireUserTranscript bool) int { + t.Helper() + seenUser := !requireUserTranscript + seenFinal, seenAudioDone, audioBytes := false, false, 0 + receipts := map[string]bool{} + for !seenUser || !seenFinal || !seenAudioDone { + event, err := conn.Recv(ctx) + saCheck(t, err) + realtimeLiveError(t, event) + switch event.Type { + case "transcript.final": + var p forward.RealtimeTranscriptPayload + saCheck(t, event.DecodePayload(&p)) + if strings.TrimSpace(p.Text) != "" { + if p.Role == "assistant" { + seenFinal = true + } + if p.Role == "user" { + seenUser = true + } + } + case "audio.delta": + pcm, err := event.Audio() + saCheck(t, err) + audioBytes += len(pcm) + key := event.WorkID + "/" + event.AnnouncementID + if !receipts[key] { + receipts[key] = true + saCheck(t, conn.PlaybackCancelled(ctx, forward.RealtimePlaybackReceipt{WorkID: event.WorkID, AnnouncementID: event.AnnouncementID})) + } + case "audio.done": + seenAudioDone = true + } + } + if audioBytes == 0 { + t.Fatal("response produced no decodable PCM audio") + } + return audioBytes +} + +func realtimeLiveError(t *testing.T, event forward.RealtimeEvent) { + t.Helper() + if event.Type == "error" { + var p forward.RealtimeErrorPayload + if event.DecodePayload(&p) != nil { + t.Fatal("malformed server error event") + } + t.Fatalf("Realtime server error code=%s conversation_id=%s event_id=%s", p.Code, event.ConversationID, event.EventID) + } + if event.Type == "voice.replaced" { + t.Fatalf("Realtime connection replaced conversation_id=%s", event.ConversationID) + } +} diff --git a/integration/sa_config_test.go b/integration/sa_config_test.go index 1b49807..0c1a532 100644 --- a/integration/sa_config_test.go +++ b/integration/sa_config_test.go @@ -12,7 +12,7 @@ import ( "testing" ) -func saEnvironment(path string, environment map[string]string, execution bool) (map[string]string, error) { +func liveEnvironment(path string, environment map[string]string) (map[string]string, error) { values := map[string]string{} file, err := os.Open(path) if err != nil && !os.IsNotExist(err) { @@ -50,6 +50,14 @@ func saEnvironment(path string, environment map[string]string, execution bool) ( for key, value := range environment { values[key] = value } + return values, nil +} + +func saEnvironment(path string, environment map[string]string, execution bool) (map[string]string, error) { + values, err := liveEnvironment(path, environment) + if err != nil { + return nil, err + } for _, name := range []string{"QODER_SA_CN_KEY", "QODER_SA_GLOBAL_KEY"} { if strings.TrimSpace(values[name]) == "" { return nil, fmt.Errorf("%s is required", name) diff --git a/internal/testsupport/contracts.go b/internal/testsupport/contracts.go index 94a0c80..00df81d 100644 --- a/internal/testsupport/contracts.go +++ b/internal/testsupport/contracts.go @@ -152,6 +152,11 @@ func call(t *testing.T, method reflect.Value, ctx context.Context, emptyPath int v.SetString("segment /?%#") } if typ.Kind() == reflect.Struct { + if typ.Name() == "RealtimeConversationNewParams" { + if err := json.Unmarshal([]byte(`"failure-contract"`), v.FieldByName("IdempotencyKey").Addr().Interface()); err != nil { + t.Fatal(err) + } + } if len(bodies) > 0 { if err := json.Unmarshal(bodies[0], v.Addr().Interface()); err != nil { t.Fatalf("request parameters: %v", err)