From b542ad6f62f72fe895db0a2e9d0284be59655a92 Mon Sep 17 00:00:00 2001 From: WANG Qingmin <75425799+FriendlyPasser@users.noreply.github.com> Date: Mon, 14 Sep 2026 18:39:29 +0800 Subject: [PATCH 1/2] fix(sessions): expose validated turn detection across SDKs Signed-off-by: WANG Qingmin <75425799+FriendlyPasser@users.noreply.github.com> --- control-plane/internal/cli/session.go | 5 + control-plane/internal/cli/session_test.go | 2 + control-plane/internal/handlers/sessions.go | 86 ++++++++--- .../internal/handlers/sessions_test.go | 126 ++++++++++++++++ .../pkg/types/session_turn_detection.go | 120 ++++++++++++++++ .../pkg/types/session_turn_detection_test.go | 77 ++++++++++ control-plane/pkg/types/types.go | 24 ++-- docs/session-turn-detection.md | 134 ++++++++++++++++++ sdk/go/README.md | 5 + sdk/go/agent/session.go | 32 +++-- sdk/go/agent/session_test.go | 38 ++++- sdk/go/agent/session_turn_detection.go | 80 +++++++++++ sdk/go/agent/session_turn_detection_test.go | 77 ++++++++++ sdk/python/README.md | 5 + sdk/python/agentfield/__init__.py | 5 + sdk/python/agentfield/agent.py | 6 +- .../agentfield/session_turn_detection.py | 72 ++++++++++ sdk/python/agentfield/sessions.py | 10 ++ sdk/python/tests/test_agent_session.py | 56 ++++++++ sdk/typescript/README.md | 5 + sdk/typescript/src/index.ts | 1 + sdk/typescript/src/session.ts | 4 + sdk/typescript/src/sessionTurnDetection.ts | 64 +++++++++ sdk/typescript/tests/agent.test.ts | 4 + .../tests/session_turn_detection.test.ts | 46 ++++++ 25 files changed, 1042 insertions(+), 42 deletions(-) create mode 100644 control-plane/pkg/types/session_turn_detection.go create mode 100644 control-plane/pkg/types/session_turn_detection_test.go create mode 100644 docs/session-turn-detection.md create mode 100644 sdk/go/agent/session_turn_detection.go create mode 100644 sdk/go/agent/session_turn_detection_test.go create mode 100644 sdk/python/agentfield/session_turn_detection.py create mode 100644 sdk/typescript/src/sessionTurnDetection.ts create mode 100644 sdk/typescript/tests/session_turn_detection.test.ts diff --git a/control-plane/internal/cli/session.go b/control-plane/internal/cli/session.go index fb63ebae8..99fefaaed 100644 --- a/control-plane/internal/cli/session.go +++ b/control-plane/internal/cli/session.go @@ -31,6 +31,7 @@ type sessionToolOptions struct { } type sessionOfferOptions struct { + target string provider string transport string sdpSource string @@ -113,6 +114,7 @@ func newSessionOfferCommand() *cobra.Command { } cmd.Flags().StringVar(&opts.provider, "provider", "", "Explicit session provider") cmd.Flags().StringVar(&opts.transport, "transport", "", "Explicit session transport") + cmd.Flags().StringVar(&opts.target, "target", "", "Registered . whose turn detection settings to use") cmd.Flags().StringVar(&opts.sdpSource, "sdp", "", "SDP offer as inline text, @path, or - for stdin; defaults to stdin") cmd.Flags().StringVarP(&opts.outputFormat, "output", "o", "raw", "Output format: raw, json, pretty, yaml") return cmd @@ -133,6 +135,9 @@ func runSessionOffer(ctx context.Context, sessionID string, opts *sessionOfferOp if strings.TrimSpace(opts.transport) != "" { values.Set("transport", opts.transport) } + if strings.TrimSpace(opts.target) != "" { + values.Set("target", opts.target) + } path := "/api/v1/session-instances/" + url.PathEscape(sessionID) + "/realtime-offer" if encoded := values.Encode(); encoded != "" { path += "?" + encoded diff --git a/control-plane/internal/cli/session_test.go b/control-plane/internal/cli/session_test.go index af6e9fe2c..dd41cb7aa 100644 --- a/control-plane/internal/cli/session_test.go +++ b/control-plane/internal/cli/session_test.go @@ -77,6 +77,7 @@ func TestRunSessionOfferPostsSDPAndWritesRawAnswer(t *testing.T) { require.Equal(t, "/api/v1/session-instances/sess-1/realtime-offer", r.URL.Path) require.Equal(t, "openai", r.URL.Query().Get("provider")) require.Equal(t, "webrtc", r.URL.Query().Get("transport")) + require.Equal(t, "support.voice", r.URL.Query().Get("target")) gotContentType = r.Header.Get("Content-Type") gotAPIKey = r.Header.Get("X-API-Key") body, err := io.ReadAll(r.Body) @@ -90,6 +91,7 @@ func TestRunSessionOfferPostsSDPAndWritesRawAnswer(t *testing.T) { var stdout bytes.Buffer err := runSessionOffer(context.Background(), "sess-1", &sessionOfferOptions{ provider: "openai", + target: "support.voice", transport: "webrtc", sdpSource: "v=0\r\noffer\r\n", outputFormat: "raw", diff --git a/control-plane/internal/handlers/sessions.go b/control-plane/internal/handlers/sessions.go index 2368f4376..bef9a1d38 100644 --- a/control-plane/internal/handlers/sessions.go +++ b/control-plane/internal/handlers/sessions.go @@ -77,29 +77,48 @@ func StartSessionHandler(store storage.StorageProvider) gin.HandlerFunc { return } + turnDetection, err := types.ParseSessionTurnDetection(capability.Provider, capability.Transport, definition.TurnDetection) + if err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) + return + } + sessionID := "sess_" + time.Now().UTC().Format("20060102_150405") + "_" + shortRandom() model := firstNonEmptySession(req.Model, definition.Model) voice := firstNonEmptySession(req.Voice, definition.Voice) + // The offer endpoint is stateless. Carry the registered target in its + // returned URL so the next request can resolve and validate its config. + offerQuery := url.Values{ + "target": {nodeID + "." + sessionName}, + "provider": {capability.Provider}, + "transport": {capability.Transport}, + } + if model != "" { + offerQuery.Set("model", model) + } + if voice != "" { + offerQuery.Set("voice", voice) + } c.JSON(http.StatusCreated, gin.H{ - "session_id": sessionID, - "target": nodeID + "." + sessionName, - "provider": capability.Provider, - "transport": capability.Transport, - "model": model, - "voice": voice, - "modalities": definition.Modalities, - "tags": definition.ApprovedTags, - "tool_targets": sessionToolTargets(nodeID, definition.Tools), - "offer_url": fmt.Sprintf("/api/v1/session-instances/%s/realtime-offer", url.PathEscape(sessionID)), - "tool_url": fmt.Sprintf("/api/v1/session-instances/%s/tools/{tool}", url.PathEscape(sessionID)), - "created_at": time.Now().UTC().Format(time.RFC3339Nano), + "turn_detection": turnDetection, + "session_id": sessionID, + "target": nodeID + "." + sessionName, + "provider": capability.Provider, + "transport": capability.Transport, + "model": model, + "voice": voice, + "modalities": definition.Modalities, + "tags": definition.ApprovedTags, + "tool_targets": sessionToolTargets(nodeID, definition.Tools), + "offer_url": fmt.Sprintf("/api/v1/session-instances/%s/realtime-offer", url.PathEscape(sessionID)) + "?" + offerQuery.Encode(), + "tool_url": fmt.Sprintf("/api/v1/session-instances/%s/tools/{tool}", url.PathEscape(sessionID)), + "created_at": time.Now().UTC().Format(time.RFC3339Nano), }) } } func SessionRealtimeOfferHandler(store storage.StorageProvider) gin.HandlerFunc { return func(c *gin.Context) { - _ = store provider := strings.TrimSpace(c.Query("provider")) transport := strings.TrimSpace(c.Query("transport")) if provider == "" || transport == "" { @@ -120,6 +139,33 @@ func SessionRealtimeOfferHandler(store storage.StorageProvider) gin.HandlerFunc c.JSON(http.StatusBadRequest, gin.H{"error": "webrtc realtime offers currently require provider=openai"}) return } + var rawTurnDetection json.RawMessage + model, voice := c.Query("model"), c.Query("voice") + if _, supplied := c.Request.URL.Query()["target"]; supplied { + target := c.Query("target") + nodeID, sessionName, ok := splitSessionTarget(target) + if !ok { + c.JSON(http.StatusBadRequest, gin.H{"error": "session target must be ."}) + return + } + definition, found := lookupSessionDefinition(c, store, nodeID, sessionName) + if !found { + return + } + if types.NormalizeSessionTransportValue(definition.Provider) != "openai" || + types.NormalizeSessionTransportValue(definition.Transport) != "webrtc" { + c.JSON(http.StatusBadRequest, gin.H{"error": "registered session must use provider=openai transport=webrtc for realtime offers"}) + return + } + rawTurnDetection = definition.TurnDetection + model = firstNonEmptySession(model, definition.Model) + voice = firstNonEmptySession(voice, definition.Voice) + } + turnDetection, err := types.ParseSessionTurnDetection("openai", "webrtc", rawTurnDetection) + if err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) + return + } if strings.TrimSpace(os.Getenv("OPENAI_API_KEY")) == "" { c.JSON(http.StatusBadGateway, gin.H{"error": "OPENAI_API_KEY is required for provider=openai transport=webrtc"}) return @@ -133,8 +179,9 @@ func SessionRealtimeOfferHandler(store storage.StorageProvider) gin.HandlerFunc c.Request.Context(), sessionPathID(c), string(sdp), - firstNonEmptySession(c.Query("model"), "gpt-realtime-2"), - firstNonEmptySession(c.Query("voice"), "marin"), + firstNonEmptySession(model, "gpt-realtime-2"), + firstNonEmptySession(voice, "marin"), + turnDetection, ) if err != nil { c.JSON(http.StatusBadGateway, gin.H{ @@ -269,7 +316,7 @@ func sessionToolTargets(nodeID string, tools []string) map[string]string { return targets } -func createOpenAIRealtimeCall(ctx context.Context, sessionID string, sdp string, model string, voice string) (string, error) { +func createOpenAIRealtimeCall(ctx context.Context, sessionID string, sdp string, model string, voice string, turnDetection *types.TurnDetection) (string, error) { var body bytes.Buffer writer := multipart.NewWriter(&body) if err := writer.WriteField("sdp", sdp); err != nil { @@ -279,8 +326,11 @@ func createOpenAIRealtimeCall(ctx context.Context, sessionID string, sdp string, "type": "realtime", "model": model, "instructions": "You are a realtime voice front end for an AgentField session. Use registered tools to route agent work through the AgentField control plane.", - "audio": map[string]interface{}{"output": map[string]interface{}{"voice": voice}}, - "tool_choice": "auto", + "audio": map[string]interface{}{ + "input": map[string]interface{}{"turn_detection": turnDetection}, + "output": map[string]interface{}{"voice": voice}, + }, + "tool_choice": "auto", } sessionBytes, _ := json.Marshal(sessionConfig) if err := writer.WriteField("session", string(sessionBytes)); err != nil { diff --git a/control-plane/internal/handlers/sessions_test.go b/control-plane/internal/handlers/sessions_test.go index 62ee09db9..551d41b71 100644 --- a/control-plane/internal/handlers/sessions_test.go +++ b/control-plane/internal/handlers/sessions_test.go @@ -6,6 +6,7 @@ import ( "io" "net/http" "net/http/httptest" + "net/url" "strings" "testing" "time" @@ -277,6 +278,10 @@ func TestSessionRealtimeOfferHandlerCallsRealtimeProvider(t *testing.T) { require.Len(t, gotSafetyID, 32) require.Contains(t, gotSession, `"model":"gpt-test"`) require.Contains(t, gotSession, `"voice":"cedar"`) + var config map[string]interface{} + require.NoError(t, json.Unmarshal([]byte(gotSession), &config)) + input := config["audio"].(map[string]interface{})["input"].(map[string]interface{}) + require.Equal(t, true, input["turn_detection"].(map[string]interface{})["interrupt_response"]) } func TestSessionRealtimeOfferHandlerSurfacesProviderErrors(t *testing.T) { @@ -403,3 +408,124 @@ func sessionTestAgent() *types.AgentNode { }}, } } + +// Exercise the complete metadata -> start -> offer -> provider boundary, rather +// than just checking that a new field exists in the registration response. +func TestSessionTurnDetectionReachesProvider(t *testing.T) { + gin.SetMode(gin.TestMode) + t.Setenv("OPENAI_API_KEY", "test-key") + original := http.DefaultClient.Transport + t.Cleanup(func() { http.DefaultClient.Transport = original }) + for _, tc := range []struct{ name, config, expected string }{ + {"legacy defaults", "", `{"type":"server_vad","threshold":0.5,"prefix_padding_ms":300,"silence_duration_ms":500,"create_response":true,"interrupt_response":true}`}, + {"custom server", `{"type":"server_vad","threshold":0,"prefix_padding_ms":0,"silence_duration_ms":750,"create_response":false,"interrupt_response":false}`, `{"type":"server_vad","threshold":0,"prefix_padding_ms":0,"silence_duration_ms":750,"create_response":false,"interrupt_response":false}`}, + {"semantic", `{"type":"semantic_vad","eagerness":"low"}`, `{"type":"semantic_vad","eagerness":"low","create_response":true,"interrupt_response":true}`}, + } { + t.Run(tc.name, func(t *testing.T) { + agent := sessionTestAgent() + raw := agent.Metadata.Custom["sessions"].([]interface{})[0].(map[string]interface{}) + raw["model"], raw["voice"] = "gpt-vad-test", "cedar" + if tc.config != "" { + raw["turn_detection"] = json.RawMessage(tc.config) + } + store := &nodeRESTStorageStub{agent: agent} + router := gin.New() + router.POST("/api/v1/session-targets/:target/start", StartSessionHandler(store)) + router.POST("/api/v1/session-instances/:session_id/realtime-offer", SessionRealtimeOfferHandler(store)) + start := httptest.NewRecorder() + router.ServeHTTP(start, httptest.NewRequest(http.MethodPost, "/api/v1/session-targets/support.voice/start", strings.NewReader(`{}`))) + require.Equal(t, http.StatusCreated, start.Code, start.Body.String()) + var result struct { + OfferURL string `json:"offer_url"` + TurnDetection json.RawMessage `json:"turn_detection"` + } + require.NoError(t, json.Unmarshal(start.Body.Bytes(), &result)) + require.JSONEq(t, tc.expected, string(result.TurnDetection)) + calls := 0 + http.DefaultClient.Transport = roundTripFunc(func(req *http.Request) (*http.Response, error) { + calls++ + require.NoError(t, req.ParseMultipartForm(1<<20)) + require.Equal(t, "v=0\r\noffer\r\n", req.FormValue("sdp")) + var config struct { + Model string `json:"model"` + Audio struct { + Output struct { + Voice string `json:"voice"` + } `json:"output"` + Input struct { + TurnDetection json.RawMessage `json:"turn_detection"` + } `json:"input"` + } `json:"audio"` + } + require.NoError(t, json.Unmarshal([]byte(req.FormValue("session")), &config)) + require.JSONEq(t, tc.expected, string(config.Audio.Input.TurnDetection)) + require.Equal(t, "gpt-vad-test", config.Model) + require.Equal(t, "cedar", config.Audio.Output.Voice) + return &http.Response{StatusCode: http.StatusOK, Header: make(http.Header), Body: io.NopCloser(strings.NewReader("answer"))}, nil + }) + if tc.name == "semantic" { + // CLI offers identify the target without repeating model/voice. + offerURL, err := url.Parse(result.OfferURL) + require.NoError(t, err) + query := offerURL.Query() + query.Del("model") + query.Del("voice") + offerURL.RawQuery = query.Encode() + result.OfferURL = offerURL.String() + } + offer := httptest.NewRecorder() + router.ServeHTTP(offer, httptest.NewRequest(http.MethodPost, result.OfferURL, strings.NewReader("v=0\r\noffer\r\n"))) + require.Equal(t, http.StatusOK, offer.Code, offer.Body.String()) + require.Equal(t, 1, calls) + }) + } +} + +func TestStartSessionRejectsInvalidTurnDetection(t *testing.T) { + for _, config := range []string{ + `{}`, `{"Type":"server_vad"}`, `{"type":"semantic_vad","eagerness":""}`, `{"type":"client_vad"}`, `{"type":"server_vad","threshold":2}`, + `{"type":"server_vad","silence_duration_ms":-1}`, `{"type":"server_vad","prefix_padding_ms":1.5}`, + `{"type":"server_vad","create_response":"false"}`, `{"type":"server_vad","interrupt_response":null}`, + `{"type":"semantic_vad","threshold":0.5}`, `{"type":"server_vad","eagerness":"low"}`, + `{"type":"semantic_vad","eagerness":"urgent"}`, `{"type":"server_vad","unknown":true}`, `[]`, + } { + t.Run(config, func(t *testing.T) { + agent := sessionTestAgent() + agent.Metadata.Custom["sessions"].([]interface{})[0].(map[string]interface{})["turn_detection"] = json.RawMessage(config) + router := gin.New() + router.POST("/:target/start", StartSessionHandler(&nodeRESTStorageStub{agent: agent})) + rec := httptest.NewRecorder() + router.ServeHTTP(rec, httptest.NewRequest(http.MethodPost, "/support.voice/start", strings.NewReader(`{}`))) + require.Equal(t, http.StatusBadRequest, rec.Code, rec.Body.String()) + require.Contains(t, rec.Body.String(), "turn_detection") + }) + } +} + +func TestSessionOfferRevalidatesRegisteredTurnDetection(t *testing.T) { + t.Setenv("OPENAI_API_KEY", "") // Invalid config must fail before credentials/upstream. + agent := sessionTestAgent() + agent.Metadata.Custom["sessions"].([]interface{})[0].(map[string]interface{})["turn_detection"] = map[string]interface{}{ + "type": "semantic_vad", "silence_duration_ms": 500, + } + router := gin.New() + router.POST("/:target/realtime-offer", SessionRealtimeOfferHandler(&nodeRESTStorageStub{agent: agent})) + rec := httptest.NewRecorder() + router.ServeHTTP(rec, httptest.NewRequest(http.MethodPost, + "/sess-1/realtime-offer?provider=openai&transport=webrtc&target=support.voice", strings.NewReader("v=0"))) + require.Equal(t, http.StatusBadRequest, rec.Code) + require.Contains(t, rec.Body.String(), "turn_detection") +} + +func TestStartSessionRejectsTurnDetectionForOpenRouter(t *testing.T) { + agent := sessionTestAgent() + raw := agent.Metadata.Custom["sessions"].([]interface{})[0].(map[string]interface{}) + raw["provider"], raw["transport"] = "openrouter", "audio_turns" + raw["turn_detection"] = map[string]interface{}{"type": "server_vad"} + router := gin.New() + router.POST("/:target/start", StartSessionHandler(&nodeRESTStorageStub{agent: agent})) + rec := httptest.NewRecorder() + router.ServeHTTP(rec, httptest.NewRequest(http.MethodPost, "/support.voice/start", strings.NewReader(`{}`))) + require.Equal(t, http.StatusBadRequest, rec.Code) + require.Contains(t, rec.Body.String(), "turn_detection requires") +} diff --git a/control-plane/pkg/types/session_turn_detection.go b/control-plane/pkg/types/session_turn_detection.go new file mode 100644 index 000000000..516115753 --- /dev/null +++ b/control-plane/pkg/types/session_turn_detection.go @@ -0,0 +1,120 @@ +package types + +import ( + "bytes" + "encoding/json" + "fmt" + "math" +) + +// TurnDetection configures OpenAI Realtime input audio. Nil options use defaults. +// Pointer fields preserve explicit false and zero values during JSON serialization. +type TurnDetection struct { + Type string `json:"type"` + Threshold *float64 `json:"threshold,omitempty"` + PrefixPaddingMS *int `json:"prefix_padding_ms,omitempty"` + SilenceDurationMS *int `json:"silence_duration_ms,omitempty"` + CreateResponse *bool `json:"create_response,omitempty"` + InterruptResponse *bool `json:"interrupt_response,omitempty"` + Eagerness string `json:"eagerness,omitempty"` +} + +// NormalizeTurnDetection validates options and returns an independent config with +// automatic responses and barge-in enabled unless explicitly disabled. +func NormalizeTurnDetection(provider, transport string, config *TurnDetection) (*TurnDetection, error) { + if provider != "openai" || (transport != "webrtc" && transport != "websocket") { + if config != nil { + return nil, fmt.Errorf("turn_detection requires provider=openai and transport=webrtc or websocket") + } + return nil, nil + } + result := TurnDetection{Type: "server_vad"} + if config != nil { + result = *config + } + switch result.Type { + case "server_vad": + if result.Eagerness != "" { + return nil, fmt.Errorf("turn_detection.eagerness is unsupported for server_vad") + } + threshold := 0.5 + if result.Threshold != nil { + threshold = *result.Threshold + } + if math.IsNaN(threshold) || math.IsInf(threshold, 0) || threshold < 0 || threshold > 1 { + return nil, fmt.Errorf("turn_detection.threshold must be a finite number between 0 and 1") + } + padding, silence := 300, 500 + if result.PrefixPaddingMS != nil { + padding = *result.PrefixPaddingMS + } + if result.SilenceDurationMS != nil { + silence = *result.SilenceDurationMS + } + if padding < 0 || silence < 0 { + return nil, fmt.Errorf("turn_detection durations must be non-negative integers") + } + result.Threshold, result.PrefixPaddingMS, result.SilenceDurationMS = &threshold, &padding, &silence + case "semantic_vad": + if result.Threshold != nil || result.PrefixPaddingMS != nil || result.SilenceDurationMS != nil { + return nil, fmt.Errorf("turn_detection threshold and durations are unsupported for semantic_vad") + } + if result.Eagerness == "" { + result.Eagerness = "auto" + } + switch result.Eagerness { + case "auto", "low", "medium", "high": + default: + return nil, fmt.Errorf("turn_detection.eagerness must be auto, low, medium, or high") + } + default: + return nil, fmt.Errorf("turn_detection.type must be server_vad or semantic_vad") + } + create, interrupt := true, true + if result.CreateResponse != nil { + create = *result.CreateResponse + } + if result.InterruptResponse != nil { + interrupt = *result.InterruptResponse + } + result.CreateResponse, result.InterruptResponse = &create, &interrupt + return &result, nil +} + +// ParseSessionTurnDetection validates untrusted registration metadata before any +// provider request. Raw JSON preserves unknown fields so they cannot be ignored. +func ParseSessionTurnDetection(provider, transport string, raw json.RawMessage) (*TurnDetection, error) { + raw = bytes.TrimSpace(raw) + if len(raw) == 0 || bytes.Equal(raw, []byte("null")) { + return NormalizeTurnDetection(provider, transport, nil) + } + var fields map[string]json.RawMessage + if err := json.Unmarshal(raw, &fields); err != nil { + return nil, fmt.Errorf("turn_detection must be an object: %w", err) + } + var config TurnDetection + decoder := json.NewDecoder(bytes.NewReader(raw)) + decoder.DisallowUnknownFields() + if err := decoder.Decode(&config); err != nil { + return nil, fmt.Errorf("invalid turn_detection: %w", err) + } + for key, value := range fields { + switch key { + case "type", "threshold", "prefix_padding_ms", "silence_duration_ms", "create_response", "interrupt_response", "eagerness": + default: + return nil, fmt.Errorf("unknown turn_detection field %q", key) + } + if bytes.Equal(bytes.TrimSpace(value), []byte("null")) { + return nil, fmt.Errorf("turn_detection.%s must not be null", key) + } + } + if _, ok := fields["eagerness"]; ok { + if config.Type != "semantic_vad" { + return nil, fmt.Errorf("turn_detection.eagerness is unsupported for %s", config.Type) + } + if config.Eagerness == "" { + return nil, fmt.Errorf("turn_detection.eagerness must be auto, low, medium, or high") + } + } + return NormalizeTurnDetection(provider, transport, &config) +} diff --git a/control-plane/pkg/types/session_turn_detection_test.go b/control-plane/pkg/types/session_turn_detection_test.go new file mode 100644 index 000000000..171dbc7be --- /dev/null +++ b/control-plane/pkg/types/session_turn_detection_test.go @@ -0,0 +1,77 @@ +package types + +import ( + "encoding/json" + "math" + "testing" +) + +func turnDetectionTestPtr[T any](v T) *T { return &v } + +func TestNormalizeTurnDetectionDefaultsAndOverrides(t *testing.T) { + for _, transport := range []string{"webrtc", "websocket"} { + config, err := NormalizeTurnDetection("openai", transport, nil) + if err != nil { + t.Fatal(err) + } + if config.Type != "server_vad" || *config.Threshold != 0.5 || *config.PrefixPaddingMS != 300 || + *config.SilenceDurationMS != 500 || !*config.CreateResponse || !*config.InterruptResponse { + t.Fatalf("unexpected defaults: %+v", config) + } + } + input := &TurnDetection{Type: "server_vad", Threshold: turnDetectionTestPtr(0.0), + PrefixPaddingMS: turnDetectionTestPtr(0), SilenceDurationMS: turnDetectionTestPtr(750), + CreateResponse: turnDetectionTestPtr(false), InterruptResponse: turnDetectionTestPtr(false)} + config, err := NormalizeTurnDetection("openai", "webrtc", input) + if err != nil { + t.Fatal(err) + } + *input.Threshold = 1 + *input.InterruptResponse = true + serialized, err := json.Marshal(config) + if err != nil { + t.Fatal(err) + } + expected := `{"type":"server_vad","threshold":0,"prefix_padding_ms":0,"silence_duration_ms":750,"create_response":false,"interrupt_response":false}` + if string(serialized) != expected { + t.Fatalf("got %s, want %s", serialized, expected) + } + semantic, err := NormalizeTurnDetection("openai", "webrtc", &TurnDetection{Type: "semantic_vad"}) + if err != nil { + t.Fatal(err) + } + serialized, err = json.Marshal(semantic) + if err != nil { + t.Fatal(err) + } + expected = `{"type":"semantic_vad","create_response":true,"interrupt_response":true,"eagerness":"auto"}` + if string(serialized) != expected { + t.Fatalf("got %s, want %s", serialized, expected) + } +} + +func TestNormalizeTurnDetectionRejectsInvalidOptions(t *testing.T) { + for _, input := range []TurnDetection{ + {}, {Type: "client_vad"}, + {Type: "server_vad", Threshold: turnDetectionTestPtr(1.1)}, + {Type: "server_vad", Threshold: turnDetectionTestPtr(math.NaN())}, + {Type: "server_vad", Threshold: turnDetectionTestPtr(math.Inf(1))}, + {Type: "server_vad", PrefixPaddingMS: turnDetectionTestPtr(-1)}, + {Type: "server_vad", SilenceDurationMS: turnDetectionTestPtr(-1)}, + {Type: "server_vad", Eagerness: "low"}, + {Type: "semantic_vad", Threshold: turnDetectionTestPtr(0.0)}, + {Type: "semantic_vad", PrefixPaddingMS: turnDetectionTestPtr(0)}, + {Type: "semantic_vad", SilenceDurationMS: turnDetectionTestPtr(0)}, + {Type: "semantic_vad", Eagerness: "urgent"}, + } { + if _, err := NormalizeTurnDetection("openai", "webrtc", &input); err == nil { + t.Fatalf("accepted invalid config: %+v", input) + } + } + if _, err := NormalizeTurnDetection("openrouter", "audio_turns", &TurnDetection{Type: "server_vad"}); err == nil { + t.Fatal("accepted VAD for openrouter") + } + if config, err := NormalizeTurnDetection("openrouter", "audio_turns", nil); err != nil || config != nil { + t.Fatalf("changed openrouter defaults: %+v, %v", config, err) + } +} diff --git a/control-plane/pkg/types/types.go b/control-plane/pkg/types/types.go index fbad40692..d1f726bb6 100644 --- a/control-plane/pkg/types/types.go +++ b/control-plane/pkg/types/types.go @@ -248,17 +248,19 @@ type SkillDefinition struct { // SessionDefinition defines a realtime session ingress provided by an agent node. type SessionDefinition struct { - Name string `json:"name"` - Provider string `json:"provider"` - Transport string `json:"transport"` - Model string `json:"model,omitempty"` - Modalities []string `json:"modalities,omitempty"` - Voice string `json:"voice,omitempty"` - Tools []string `json:"tools,omitempty"` - Metadata map[string]interface{} `json:"metadata,omitempty"` - Tags []string `json:"tags,omitempty"` - ProposedTags []string `json:"proposed_tags,omitempty"` - ApprovedTags []string `json:"approved_tags,omitempty"` + // Keep raw options until validation so unknown or malformed fields are rejected explicitly. + TurnDetection json.RawMessage `json:"turn_detection,omitempty"` + Name string `json:"name"` + Provider string `json:"provider"` + Transport string `json:"transport"` + Model string `json:"model,omitempty"` + Modalities []string `json:"modalities,omitempty"` + Voice string `json:"voice,omitempty"` + Tools []string `json:"tools,omitempty"` + Metadata map[string]interface{} `json:"metadata,omitempty"` + Tags []string `json:"tags,omitempty"` + ProposedTags []string `json:"proposed_tags,omitempty"` + ApprovedTags []string `json:"approved_tags,omitempty"` } // HydrateAgentSessions copies session definitions from metadata.custom.sessions diff --git a/docs/session-turn-detection.md b/docs/session-turn-detection.md new file mode 100644 index 000000000..5322a66da --- /dev/null +++ b/docs/session-turn-detection.md @@ -0,0 +1,134 @@ +# Session turn detection and interruption + +OpenAI sessions accept `turn_detection` in Python and TypeScript, and +`WithSessionTurnDetection` in Go. It is supported with explicit +`provider="openai"` and `transport="webrtc"` or `"websocket"`. Supplying it for +OpenRouter `audio_turns` is an error; AgentField does not switch providers or +transports. + +Omitting the configuration enables interruptible server VAD: + +```json +{ + "type": "server_vad", + "threshold": 0.5, + "prefix_padding_ms": 300, + "silence_duration_ms": 500, + "create_response": true, + "interrupt_response": true +} +``` + +These defaults also apply to older OpenAI registrations without this field. +An explicitly supplied object must specify `type`. Missing optional fields use +mode-specific defaults; explicit `false` and `0` are preserved. + +## Python + +```python +from agentfield import Agent + +app = Agent("support") + +@app.session( + "voice", + provider="openai", + transport="webrtc", + turn_detection={ + "type": "server_vad", + "threshold": 0.6, + "silence_duration_ms": 700, + "interrupt_response": True, + }, +) +async def voice(session): + pass +``` + +`ServerVAD`, `SemanticVAD`, and the `TurnDetection` union are exported for type +annotations. The decorator validates the dictionary when the session is declared. + +## TypeScript + +```typescript +import { Agent } from '@agentfield/sdk'; + +const app = new Agent({ nodeId: 'support' }); +app.session('voice', { + provider: 'openai', + transport: 'webrtc', + turn_detection: { + type: 'semantic_vad', + eagerness: 'low', + interrupt_response: true + } +}, async (session) => {}); +``` + +`TurnDetection` is an exported discriminated union. Runtime validation also +rejects invalid values supplied by JavaScript or external configuration. + +## Go + +```go +interrupt := false +silence := 700 +err := app.RegisterSession("voice", "openai", "webrtc", + agent.WithSessionTurnDetection(agent.TurnDetection{ + Type: "server_vad", + SilenceDurationMS: &silence, + InterruptResponse: &interrupt, + }), +) +``` + +Optional numeric and boolean fields are pointers so an unset field can be +distinguished from an explicit zero or false. `RegisterSession` returns a +validation error before updating the registry. + +## Supported options + +| Option | Modes | Default | Validation | +| --- | --- | --- | --- | +| `type` | Both | `server_vad` when config is omitted | `server_vad` or `semantic_vad`; required in an explicit object | +| `threshold` | Server | `0.5` | Finite number from 0 to 1 | +| `prefix_padding_ms` | Server | `300` | Non-negative integer milliseconds | +| `silence_duration_ms` | Server | `500` | Non-negative integer milliseconds | +| `eagerness` | Semantic | `auto` | `auto`, `low`, `medium`, or `high` | +| `create_response` | Both | `true` | Boolean; automatically respond after a detected turn | +| `interrupt_response` | Both | `true` | Boolean; interrupt an ongoing response when speech starts | + +Server-only fields cannot be supplied with semantic VAD, and `eagerness` cannot +be supplied with server VAD. Unknown fields, invalid values, and null field values +are rejected. Setting both response flags to `false` keeps speech detection +active while leaving response creation and cancellation to the client. + +## Control-plane connection + +Start the registered session with +`POST /api/v1/session-targets/./start`, then POST the raw SDP offer +to the returned `offer_url` with `Content-Type: application/sdp`. Preserve its +query parameters: they identify the registered target, provider, transport, +model, and voice. The start response also includes the resolved `turn_detection`. + +The offer endpoint re-reads the registered target and validates its configuration +before contacting OpenAI. It sends the resolved options under +`session.audio.input.turn_detection` in the multipart session configuration. +This is a stateless lookup, so a registration change between start and offer is +reflected when the offer is submitted. There is no session database migration. + +The CLI can select the same registered configuration: + +```sh +agentfield session offer --provider openai --transport webrtc \ + --target support.voice --sdp @offer.sdp +``` + +Legacy direct offers without a target keep working and use the interruptible +server-VAD defaults. Clients constructing offer URLs themselves must include +`target=.` to use author-defined settings. The existing offer +endpoint remains WebRTC-only; accepting WebSocket registration metadata does +not add a WebSocket connection adapter. + +See [OpenAI's VAD guide](https://developers.openai.com/api/docs/guides/realtime-vad) +for the provider's turn detection and interruption behavior. diff --git a/sdk/go/README.md b/sdk/go/README.md index 14033bb6f..44a0aaf5c 100644 --- a/sdk/go/README.md +++ b/sdk/go/README.md @@ -139,3 +139,8 @@ go test ./... ## License Distributed under the Apache 2.0 License. See the repository root for full details. + +## Realtime session turn detection + +See [session turn detection and interruption](../../docs/session-turn-detection.md) +for VAD configuration, defaults, validation, and the WebRTC connection flow. diff --git a/sdk/go/agent/session.go b/sdk/go/agent/session.go index 4790f60fd..5ead5494b 100644 --- a/sdk/go/agent/session.go +++ b/sdk/go/agent/session.go @@ -1,21 +1,27 @@ package agent type SessionDefinition struct { - Name string `json:"name"` - Provider string `json:"provider"` - Transport string `json:"transport"` - Model string `json:"model,omitempty"` - Modalities []string `json:"modalities"` - Voice string `json:"voice,omitempty"` - Tools []string `json:"tools"` - Tags []string `json:"tags,omitempty"` - ProposedTags []string `json:"proposed_tags,omitempty"` - ApprovedTags []string `json:"approved_tags,omitempty"` - Metadata map[string]any `json:"metadata"` + Name string `json:"name"` + Provider string `json:"provider"` + Transport string `json:"transport"` + Model string `json:"model,omitempty"` + Modalities []string `json:"modalities"` + TurnDetection *TurnDetection `json:"turn_detection,omitempty"` + Voice string `json:"voice,omitempty"` + Tools []string `json:"tools"` + Tags []string `json:"tags,omitempty"` + ProposedTags []string `json:"proposed_tags,omitempty"` + ApprovedTags []string `json:"approved_tags,omitempty"` + Metadata map[string]any `json:"metadata"` } type SessionOption func(*SessionDefinition) +// WithSessionTurnDetection sets VAD options; RegisterSession validates them. +func WithSessionTurnDetection(config TurnDetection) SessionOption { + return func(s *SessionDefinition) { s.TurnDetection = &config } +} + func WithSessionModel(model string) SessionOption { return func(s *SessionDefinition) { s.Model = model } } @@ -90,6 +96,10 @@ func (a *Agent) RegisterSession(name string, provider string, transport string, definition.Metadata = map[string]any{} } + definition.TurnDetection, err = NormalizeTurnDetection(definition.Provider, definition.Transport, definition.TurnDetection) + if err != nil { + return err + } a.sessions[name] = definition return nil } diff --git a/sdk/go/agent/session_test.go b/sdk/go/agent/session_test.go index e0e1e80db..17bb0fe43 100644 --- a/sdk/go/agent/session_test.go +++ b/sdk/go/agent/session_test.go @@ -1,6 +1,10 @@ package agent -import "testing" +import ( + "encoding/json" + "strings" + "testing" +) func TestAgentRegisterSessionStoresExplicitDefinition(t *testing.T) { a, err := New(Config{NodeID: "support", Version: "v1"}) @@ -45,3 +49,35 @@ func TestAgentRegisterSessionRejectsInvalidTransport(t *testing.T) { t.Fatal("expected invalid provider/transport error") } } + +func TestAgentRegisterSessionTurnDetection(t *testing.T) { + a, err := New(Config{NodeID: "support", Version: "v1"}) + if err != nil { + t.Fatal(err) + } + err = a.RegisterSession("voice", "OpenAI", "WebRTC", WithSessionTurnDetection(TurnDetection{ + Type: "server_vad", Threshold: turnDetectionTestPtr(0.0), InterruptResponse: turnDetectionTestPtr(false), + })) + if err != nil { + t.Fatal(err) + } + body, err := json.Marshal(a.SessionDefinitions()) + if err != nil { + t.Fatal(err) + } + for _, value := range []string{`"turn_detection":`, `"threshold":0`, `"interrupt_response":false`} { + if !strings.Contains(string(body), value) { + t.Fatalf("missing %s in %s", value, body) + } + } + err = a.RegisterSession("invalid", "openai", "webrtc", WithSessionTurnDetection(TurnDetection{ + Type: "semantic_vad", Threshold: turnDetectionTestPtr(0.5), + })) + if err == nil || len(a.SessionDefinitions()) != 1 { + t.Fatal("invalid registration must not change registry") + } + err = a.RegisterSession("invalid", "openrouter", "audio_turns", WithSessionTurnDetection(TurnDetection{Type: "server_vad"})) + if err == nil { + t.Fatal("expected unsupported provider error") + } +} diff --git a/sdk/go/agent/session_turn_detection.go b/sdk/go/agent/session_turn_detection.go new file mode 100644 index 000000000..877de51ce --- /dev/null +++ b/sdk/go/agent/session_turn_detection.go @@ -0,0 +1,80 @@ +package agent + +import ( + "fmt" + "math" +) + +// TurnDetection configures OpenAI Realtime input audio. Nil options use defaults. +// Pointer fields preserve explicit false and zero values during JSON serialization. +type TurnDetection struct { + Type string `json:"type"` + Threshold *float64 `json:"threshold,omitempty"` + PrefixPaddingMS *int `json:"prefix_padding_ms,omitempty"` + SilenceDurationMS *int `json:"silence_duration_ms,omitempty"` + CreateResponse *bool `json:"create_response,omitempty"` + InterruptResponse *bool `json:"interrupt_response,omitempty"` + Eagerness string `json:"eagerness,omitempty"` +} + +// NormalizeTurnDetection validates options and returns an independent config with +// automatic responses and barge-in enabled unless explicitly disabled. +func NormalizeTurnDetection(provider, transport string, config *TurnDetection) (*TurnDetection, error) { + if provider != "openai" || (transport != "webrtc" && transport != "websocket") { + if config != nil { + return nil, fmt.Errorf("turn_detection requires provider=openai and transport=webrtc or websocket") + } + return nil, nil + } + result := TurnDetection{Type: "server_vad"} + if config != nil { + result = *config + } + switch result.Type { + case "server_vad": + if result.Eagerness != "" { + return nil, fmt.Errorf("turn_detection.eagerness is unsupported for server_vad") + } + threshold := 0.5 + if result.Threshold != nil { + threshold = *result.Threshold + } + if math.IsNaN(threshold) || math.IsInf(threshold, 0) || threshold < 0 || threshold > 1 { + return nil, fmt.Errorf("turn_detection.threshold must be a finite number between 0 and 1") + } + padding, silence := 300, 500 + if result.PrefixPaddingMS != nil { + padding = *result.PrefixPaddingMS + } + if result.SilenceDurationMS != nil { + silence = *result.SilenceDurationMS + } + if padding < 0 || silence < 0 { + return nil, fmt.Errorf("turn_detection durations must be non-negative integers") + } + result.Threshold, result.PrefixPaddingMS, result.SilenceDurationMS = &threshold, &padding, &silence + case "semantic_vad": + if result.Threshold != nil || result.PrefixPaddingMS != nil || result.SilenceDurationMS != nil { + return nil, fmt.Errorf("turn_detection threshold and durations are unsupported for semantic_vad") + } + if result.Eagerness == "" { + result.Eagerness = "auto" + } + switch result.Eagerness { + case "auto", "low", "medium", "high": + default: + return nil, fmt.Errorf("turn_detection.eagerness must be auto, low, medium, or high") + } + default: + return nil, fmt.Errorf("turn_detection.type must be server_vad or semantic_vad") + } + create, interrupt := true, true + if result.CreateResponse != nil { + create = *result.CreateResponse + } + if result.InterruptResponse != nil { + interrupt = *result.InterruptResponse + } + result.CreateResponse, result.InterruptResponse = &create, &interrupt + return &result, nil +} diff --git a/sdk/go/agent/session_turn_detection_test.go b/sdk/go/agent/session_turn_detection_test.go new file mode 100644 index 000000000..d5af2b842 --- /dev/null +++ b/sdk/go/agent/session_turn_detection_test.go @@ -0,0 +1,77 @@ +package agent + +import ( + "encoding/json" + "math" + "testing" +) + +func turnDetectionTestPtr[T any](v T) *T { return &v } + +func TestNormalizeTurnDetectionDefaultsAndOverrides(t *testing.T) { + for _, transport := range []string{"webrtc", "websocket"} { + config, err := NormalizeTurnDetection("openai", transport, nil) + if err != nil { + t.Fatal(err) + } + if config.Type != "server_vad" || *config.Threshold != 0.5 || *config.PrefixPaddingMS != 300 || + *config.SilenceDurationMS != 500 || !*config.CreateResponse || !*config.InterruptResponse { + t.Fatalf("unexpected defaults: %+v", config) + } + } + input := &TurnDetection{Type: "server_vad", Threshold: turnDetectionTestPtr(0.0), + PrefixPaddingMS: turnDetectionTestPtr(0), SilenceDurationMS: turnDetectionTestPtr(750), + CreateResponse: turnDetectionTestPtr(false), InterruptResponse: turnDetectionTestPtr(false)} + config, err := NormalizeTurnDetection("openai", "webrtc", input) + if err != nil { + t.Fatal(err) + } + *input.Threshold = 1 + *input.InterruptResponse = true + serialized, err := json.Marshal(config) + if err != nil { + t.Fatal(err) + } + expected := `{"type":"server_vad","threshold":0,"prefix_padding_ms":0,"silence_duration_ms":750,"create_response":false,"interrupt_response":false}` + if string(serialized) != expected { + t.Fatalf("got %s, want %s", serialized, expected) + } + semantic, err := NormalizeTurnDetection("openai", "webrtc", &TurnDetection{Type: "semantic_vad"}) + if err != nil { + t.Fatal(err) + } + serialized, err = json.Marshal(semantic) + if err != nil { + t.Fatal(err) + } + expected = `{"type":"semantic_vad","create_response":true,"interrupt_response":true,"eagerness":"auto"}` + if string(serialized) != expected { + t.Fatalf("got %s, want %s", serialized, expected) + } +} + +func TestNormalizeTurnDetectionRejectsInvalidOptions(t *testing.T) { + for _, input := range []TurnDetection{ + {}, {Type: "client_vad"}, + {Type: "server_vad", Threshold: turnDetectionTestPtr(1.1)}, + {Type: "server_vad", Threshold: turnDetectionTestPtr(math.NaN())}, + {Type: "server_vad", Threshold: turnDetectionTestPtr(math.Inf(1))}, + {Type: "server_vad", PrefixPaddingMS: turnDetectionTestPtr(-1)}, + {Type: "server_vad", SilenceDurationMS: turnDetectionTestPtr(-1)}, + {Type: "server_vad", Eagerness: "low"}, + {Type: "semantic_vad", Threshold: turnDetectionTestPtr(0.0)}, + {Type: "semantic_vad", PrefixPaddingMS: turnDetectionTestPtr(0)}, + {Type: "semantic_vad", SilenceDurationMS: turnDetectionTestPtr(0)}, + {Type: "semantic_vad", Eagerness: "urgent"}, + } { + if _, err := NormalizeTurnDetection("openai", "webrtc", &input); err == nil { + t.Fatalf("accepted invalid config: %+v", input) + } + } + if _, err := NormalizeTurnDetection("openrouter", "audio_turns", &TurnDetection{Type: "server_vad"}); err == nil { + t.Fatal("accepted VAD for openrouter") + } + if config, err := NormalizeTurnDetection("openrouter", "audio_turns", nil); err != nil || config != nil { + t.Fatalf("changed openrouter defaults: %+v, %v", config, err) + } +} diff --git a/sdk/python/README.md b/sdk/python/README.md index 35bba67e7..af05356e5 100644 --- a/sdk/python/README.md +++ b/sdk/python/README.md @@ -211,3 +211,8 @@ and CI do not rely on pytest's predictable default temp directory layout. ## License Distributed under the Apache 2.0 License. See the project root `LICENSE` for details. + +## Realtime session turn detection + +See [session turn detection and interruption](../../docs/session-turn-detection.md) +for VAD configuration, defaults, validation, and the WebRTC connection flow. diff --git a/sdk/python/agentfield/__init__.py b/sdk/python/agentfield/__init__.py index f4259fba7..e5ca2c5e0 100644 --- a/sdk/python/agentfield/__init__.py +++ b/sdk/python/agentfield/__init__.py @@ -91,7 +91,12 @@ capabilities_to_tool_schemas, ) +from .session_turn_detection import ServerVAD, SemanticVAD, TurnDetection + __all__ = [ + "ServerVAD", + "SemanticVAD", + "TurnDetection", "Agent", "AgentMesh", "CostTracker", diff --git a/sdk/python/agentfield/agent.py b/sdk/python/agentfield/agent.py index 6ffea2041..ed65c777d 100644 --- a/sdk/python/agentfield/agent.py +++ b/sdk/python/agentfield/agent.py @@ -72,6 +72,7 @@ from agentfield.async_config import AsyncConfig from agentfield.async_execution_manager import AsyncExecutionManager from agentfield.pydantic_utils import convert_function_args, should_convert_args +from agentfield.session_turn_detection import TurnDetection from agentfield.sessions import ( RealtimeSession, build_session_definition, @@ -1815,6 +1816,7 @@ def session( model: Optional[str] = None, modalities: Optional[List[str]] = None, voice: Optional[str] = None, + turn_detection: Optional[TurnDetection] = None, tools: Optional[List[str]] = None, tags: Optional[List[str]] = None, metadata: Optional[Dict[str, Any]] = None, @@ -1823,7 +1825,8 @@ def session( Provider and transport are both explicit; AgentField does not infer or switch them. Unsupported combinations fail at declaration time and again - at control-plane session start. + at control-plane session start. OpenAI sessions default to interruptible + server VAD; turn_detection configures detection and response behavior. """ definition = build_session_definition( @@ -1833,6 +1836,7 @@ def session( model=model, modalities=modalities, voice=voice, + turn_detection=turn_detection, tools=tools, tags=tags, metadata=metadata, diff --git a/sdk/python/agentfield/session_turn_detection.py b/sdk/python/agentfield/session_turn_detection.py new file mode 100644 index 000000000..a564d6cb7 --- /dev/null +++ b/sdk/python/agentfield/session_turn_detection.py @@ -0,0 +1,72 @@ +"""Validated OpenAI Realtime turn detection options.""" + +import math +from typing import Any, Dict, Literal, Optional, TypedDict, Union, cast + + +class _ServerVADType(TypedDict): + type: Literal["server_vad"] + + +class ServerVAD(_ServerVADType, total=False): + threshold: float + prefix_padding_ms: int + silence_duration_ms: int + create_response: bool + interrupt_response: bool + + +class _SemanticVADType(TypedDict): + type: Literal["semantic_vad"] + + +class SemanticVAD(_SemanticVADType, total=False): + eagerness: Literal["auto", "low", "medium", "high"] + create_response: bool + interrupt_response: bool + + +TurnDetection = Union[ServerVAD, SemanticVAD] + + +def normalize_turn_detection( + provider: str, transport: str, config: Optional[TurnDetection] +) -> Optional[TurnDetection]: + """Default to interruptible server VAD; never infer a supplied config's type.""" + if provider != "openai" or transport not in ("webrtc", "websocket"): + if config is not None: + raise ValueError("turn_detection requires provider=openai and transport=webrtc or websocket") + return None + if config is None: + config = {"type": "server_vad"} + if not isinstance(config, dict): + raise ValueError("turn_detection must be an object") + kind = config.get("type") + common = {"type", "create_response", "interrupt_response"} + if kind == "server_vad": + allowed = common | {"threshold", "prefix_padding_ms", "silence_duration_ms"} + elif kind == "semantic_vad": + allowed = common | {"eagerness"} + else: + raise ValueError("turn_detection.type must be server_vad or semantic_vad") + for key, value in config.items(): + if key not in allowed: + raise ValueError(f"turn_detection.{key} is unsupported for {kind}") + if key in ("create_response", "interrupt_response") and type(value) is not bool: + raise ValueError(f"turn_detection.{key} must be a boolean") + if key == "threshold" and ( + type(value) not in (int, float) or not 0 <= value <= 1 or not math.isfinite(value) + ): + raise ValueError("turn_detection.threshold must be a finite number between 0 and 1") + if key in ("prefix_padding_ms", "silence_duration_ms") and ( + type(value) is not int or value < 0 + ): + raise ValueError(f"turn_detection.{key} must be a non-negative integer") + if key == "eagerness" and value not in ("auto", "low", "medium", "high"): + raise ValueError("turn_detection.eagerness must be auto, low, medium, or high") + defaults: Dict[str, Any] = {"create_response": True, "interrupt_response": True} + if kind == "server_vad": + defaults.update(threshold=0.5, prefix_padding_ms=300, silence_duration_ms=500) + else: + defaults["eagerness"] = "auto" + return cast(TurnDetection, {**defaults, **config}) diff --git a/sdk/python/agentfield/sessions.py b/sdk/python/agentfield/sessions.py index 12f3e3af0..5f658658a 100644 --- a/sdk/python/agentfield/sessions.py +++ b/sdk/python/agentfield/sessions.py @@ -6,6 +6,7 @@ from typing import Any, Awaitable, Callable, Dict, List, Optional from .session_transport import validate_session_transport +from .session_turn_detection import TurnDetection, normalize_turn_detection SessionHandler = Callable[["RealtimeSession"], Awaitable[Any]] @@ -24,6 +25,7 @@ class SessionDefinition: proposed_tags: List[str] = field(default_factory=list) approved_tags: List[str] = field(default_factory=list) metadata: Dict[str, Any] = field(default_factory=dict) + turn_detection: Optional[TurnDetection] = None def to_dict(self) -> Dict[str, Any]: return { @@ -38,6 +40,10 @@ def to_dict(self) -> Dict[str, Any]: "proposed_tags": list(self.proposed_tags or self.tags), "approved_tags": list(self.approved_tags), "metadata": dict(self.metadata), + **( + {"turn_detection": dict(self.turn_detection)} + if self.turn_detection is not None else {} + ), } @@ -93,12 +99,16 @@ def build_session_definition( tools: Optional[List[str]] = None, tags: Optional[List[str]] = None, metadata: Optional[Dict[str, Any]] = None, + turn_detection: Optional[TurnDetection] = None, ) -> SessionDefinition: capability = validate_session_transport(provider, transport) return SessionDefinition( name=name, provider=capability.provider, transport=capability.transport, + turn_detection=normalize_turn_detection( + capability.provider, capability.transport, turn_detection + ), model=model, modalities=list(modalities or ["audio", "text"]), voice=voice, diff --git a/sdk/python/tests/test_agent_session.py b/sdk/python/tests/test_agent_session.py index c6f7ddb1d..6a0289db9 100644 --- a/sdk/python/tests/test_agent_session.py +++ b/sdk/python/tests/test_agent_session.py @@ -27,6 +27,11 @@ async def voice(session): "model": "gpt-realtime-2", "modalities": ["audio", "text"], "voice": "marin", + "turn_detection": { + "type": "server_vad", "threshold": 0.5, + "prefix_padding_ms": 300, "silence_duration_ms": 500, + "create_response": True, "interrupt_response": True, + }, "tools": ["launch_support_workflow"], "tags": ["voice", "pii"], "proposed_tags": ["voice", "pii"], @@ -42,3 +47,54 @@ def test_app_session_rejects_unsupported_provider_transport_pair(): with pytest.raises(SessionTransportError): app.session("voice", provider="openrouter", transport="webrtc") + + +@pytest.mark.parametrize("transport", ["webrtc", "websocket"]) +@pytest.mark.parametrize("config,expected", [ + ({"type": "server_vad", "threshold": 0, "prefix_padding_ms": 0, + "silence_duration_ms": 750, "create_response": False, "interrupt_response": False}, + {"type": "server_vad", "threshold": 0, "prefix_padding_ms": 0, + "silence_duration_ms": 750, "create_response": False, "interrupt_response": False}), + ({"type": "semantic_vad", "eagerness": "low"}, + {"type": "semantic_vad", "eagerness": "low", "create_response": True, "interrupt_response": True}), +]) +def test_session_turn_detection_registration(transport, config, expected): + app = Agent("support", auto_register=False) + config = dict(config) + async def handler(session): + return session + app.session("voice", provider="openai", transport=transport, turn_detection=config)(handler) + config["type"] = "changed-after-registration" + assert app.sessions[0]["turn_detection"] == expected + assert app._build_agent_metadata()["sessions"][0]["turn_detection"] == expected + + +@pytest.mark.parametrize("config", [ + {}, {"type": "client_vad"}, {"type": "server_vad", "threshold": 2}, + {"type": "server_vad", "threshold": float("nan")}, + {"type": "server_vad", "threshold": True}, + {"type": "server_vad", "silence_duration_ms": -1}, + {"type": "server_vad", "prefix_padding_ms": 1.5}, + {"type": "server_vad", "create_response": "false"}, + {"type": "server_vad", "interrupt_response": None}, + {"type": "semantic_vad", "threshold": 0.5}, + {"type": "server_vad", "eagerness": "low"}, + {"type": "semantic_vad", "eagerness": "urgent"}, + {"type": "server_vad", "unknown": True}, [], +]) +def test_session_rejects_invalid_turn_detection(config): + app = Agent("support", auto_register=False) + with pytest.raises(ValueError, match="turn_detection"): + app.session("voice", provider="openai", transport="webrtc", turn_detection=config) + assert app.sessions == [] + + +def test_turn_detection_requires_openai(): + app = Agent("support", auto_register=False) + with pytest.raises(ValueError, match="turn_detection requires"): + app.session("voice", provider="openrouter", transport="audio_turns", + turn_detection={"type": "server_vad"}) + async def handler(session): + return session + app.session("voice", provider="openrouter", transport="audio_turns")(handler) + assert "turn_detection" not in app.sessions[0] diff --git a/sdk/typescript/README.md b/sdk/typescript/README.md index eb94cd9a7..1cb4efa47 100644 --- a/sdk/typescript/README.md +++ b/sdk/typescript/README.md @@ -168,3 +168,8 @@ agent.reasoner<{ task: string }, { status: string }>('deploy', async (ctx) => { **Methods:** `requestApproval()`, `getApprovalStatus()`, `waitForApproval()` See `examples/ts-node-examples/waiting-state/` for a complete working example. + +## Realtime session turn detection + +See [session turn detection and interruption](../../docs/session-turn-detection.md) +for VAD configuration, defaults, validation, and the WebRTC connection flow. diff --git a/sdk/typescript/src/index.ts b/sdk/typescript/src/index.ts index 1a87b3b52..9a4141308 100644 --- a/sdk/typescript/src/index.ts +++ b/sdk/typescript/src/index.ts @@ -30,5 +30,6 @@ export * from './usage/costTracker.js'; export * from './status/ExecutionStatus.js'; export * from './approval/ApprovalClient.js'; export * from './sessionTransport.js'; +export * from './sessionTurnDetection.js'; export * from './session.js'; export * from './triggers/index.js'; diff --git a/sdk/typescript/src/session.ts b/sdk/typescript/src/session.ts index 0987eac10..0a2999581 100644 --- a/sdk/typescript/src/session.ts +++ b/sdk/typescript/src/session.ts @@ -1,4 +1,5 @@ import { validateSessionTransport } from './sessionTransport.js'; +import { normalizeTurnDetection, type TurnDetection } from './sessionTurnDetection.js'; export interface SessionDefinition { name: string; @@ -7,6 +8,7 @@ export interface SessionDefinition { model?: string; modalities: string[]; voice?: string; + turn_detection?: TurnDetection; tools: string[]; tags: string[]; proposed_tags: string[]; @@ -20,6 +22,7 @@ export interface SessionOptions { model?: string; modalities?: string[]; voice?: string; + turn_detection?: TurnDetection; tools?: string[]; tags?: string[]; metadata?: Record; @@ -57,6 +60,7 @@ export function buildSessionDefinition(name: string, options: SessionOptions): S model: options.model, modalities: options.modalities ?? ['audio', 'text'], voice: options.voice, + turn_detection: normalizeTurnDetection(capability.provider, capability.transport, options.turn_detection), tools: options.tools ?? [], tags: options.tags ?? [], proposed_tags: options.tags ?? [], diff --git a/sdk/typescript/src/sessionTurnDetection.ts b/sdk/typescript/src/sessionTurnDetection.ts new file mode 100644 index 000000000..460d898a6 --- /dev/null +++ b/sdk/typescript/src/sessionTurnDetection.ts @@ -0,0 +1,64 @@ +export type TurnDetection = { + create_response?: boolean; + interrupt_response?: boolean; +} & ({ + type: 'server_vad'; + threshold?: number; + prefix_padding_ms?: number; + silence_duration_ms?: number; + eagerness?: never; +} | { + type: 'semantic_vad'; + eagerness?: 'auto' | 'low' | 'medium' | 'high'; + threshold?: never; + prefix_padding_ms?: never; + silence_duration_ms?: never; +}); + +export function normalizeTurnDetection( + provider: string, transport: string, config?: TurnDetection +): TurnDetection | undefined { + if (provider !== 'openai' || !['webrtc', 'websocket'].includes(transport)) { + if (config !== undefined) { + throw new Error('turn_detection requires provider=openai and transport=webrtc or websocket'); + } + return undefined; + } + if (config === undefined) config = { type: 'server_vad' }; + if (config === null || typeof config !== 'object' || Array.isArray(config)) { + throw new Error('turn_detection must be an object'); + } + const kind = config.type; + const common = ['type', 'create_response', 'interrupt_response']; + let allowed: string[]; + if (kind === 'server_vad') { + allowed = [...common, 'threshold', 'prefix_padding_ms', 'silence_duration_ms']; + } else if (kind === 'semantic_vad') { + allowed = [...common, 'eagerness']; + } else { + throw new Error('turn_detection.type must be server_vad or semantic_vad'); + } + const supplied: Record = {}; + for (const [key, value] of Object.entries(config)) { + if (!allowed.includes(key)) throw new Error(`turn_detection.${key} is unsupported for ${kind}`); + if (value === undefined) continue; + if (['create_response', 'interrupt_response'].includes(key) && typeof value !== 'boolean') { + throw new Error(`turn_detection.${key} must be a boolean`); + } + if (key === 'threshold' && (typeof value !== 'number' || !Number.isFinite(value) || value < 0 || value > 1)) { + throw new Error('turn_detection.threshold must be a finite number between 0 and 1'); + } + if (['prefix_padding_ms', 'silence_duration_ms'].includes(key) && + (typeof value !== 'number' || !Number.isSafeInteger(value) || value < 0)) { + throw new Error(`turn_detection.${key} must be a non-negative integer`); + } + if (key === 'eagerness' && !['auto', 'low', 'medium', 'high'].includes(value as string)) { + throw new Error('turn_detection.eagerness must be auto, low, medium, or high'); + } + supplied[key] = value; + } + const defaults = kind === 'server_vad' + ? { threshold: 0.5, prefix_padding_ms: 300, silence_duration_ms: 500 } + : { eagerness: 'auto' }; + return { ...defaults, create_response: true, interrupt_response: true, ...supplied } as TurnDetection; +} diff --git a/sdk/typescript/tests/agent.test.ts b/sdk/typescript/tests/agent.test.ts index 5b4596acd..63b1816cb 100644 --- a/sdk/typescript/tests/agent.test.ts +++ b/sdk/typescript/tests/agent.test.ts @@ -81,6 +81,10 @@ describe('Agent', () => { model: 'gpt-realtime-2', modalities: ['audio', 'text'], voice: 'marin', + turn_detection: { + type: 'server_vad', threshold: 0.5, prefix_padding_ms: 300, + silence_duration_ms: 500, create_response: true, interrupt_response: true + }, tools: ['support.resolve_voice_turn'], tags: [], proposed_tags: [], diff --git a/sdk/typescript/tests/session_turn_detection.test.ts b/sdk/typescript/tests/session_turn_detection.test.ts new file mode 100644 index 000000000..3aee4fc1b --- /dev/null +++ b/sdk/typescript/tests/session_turn_detection.test.ts @@ -0,0 +1,46 @@ +import { describe, expect, it } from 'vitest'; +import { Agent } from '../src/agent/Agent.js'; +import { buildSessionDefinition } from '../src/session.js'; +import type { TurnDetection } from '../src/sessionTurnDetection.js'; + +describe('session turn detection', () => { + it.each(['webrtc', 'websocket'])('registers and serializes %s options preserving false and zero', (transport) => { + const agent = new Agent({ nodeId: 'support', devMode: true }); + const config: TurnDetection = { + type: 'server_vad', threshold: 0, prefix_padding_ms: 0, silence_duration_ms: 750, + create_response: false, interrupt_response: false + }; + agent.session('voice', { provider: 'OpenAI', transport, turn_detection: config }, async () => ({})); + const expected = { ...config }; + config.threshold = 1; + expect(JSON.parse(JSON.stringify(agent.sessionDefinitions()))[0].turn_detection).toEqual(expected); + }); + + it('uses semantic defaults without server-only fields', () => { + expect(buildSessionDefinition('voice', { + provider: 'openai', transport: 'webrtc', turn_detection: { type: 'semantic_vad', eagerness: 'low' } + }).turn_detection).toEqual({ type: 'semantic_vad', eagerness: 'low', create_response: true, interrupt_response: true }); + }); + + it.each([ + {}, { type: 'client_vad' }, { type: 'server_vad', threshold: 2 }, + { type: 'server_vad', threshold: NaN }, { type: 'server_vad', threshold: true }, + { type: 'server_vad', silence_duration_ms: -1 }, { type: 'server_vad', prefix_padding_ms: 1.5 }, + { type: 'server_vad', create_response: 'false' }, { type: 'server_vad', interrupt_response: null }, + { type: 'semantic_vad', threshold: 0.5 }, { type: 'server_vad', eagerness: 'low' }, + { type: 'semantic_vad', eagerness: 'urgent' }, { type: 'server_vad', unknown: true }, [] + ])('rejects invalid runtime input %j', (config) => { + expect(() => buildSessionDefinition('voice', { + provider: 'openai', transport: 'webrtc', turn_detection: config as TurnDetection + })).toThrow(/turn_detection/); + }); + + it('rejects VAD for openrouter without changing its defaults', () => { + expect(() => buildSessionDefinition('voice', { + provider: 'openrouter', transport: 'audio_turns', turn_detection: { type: 'server_vad' } + })).toThrow(/turn_detection requires/); + expect(buildSessionDefinition('voice', { + provider: 'openrouter', transport: 'audio_turns' + }).turn_detection).toBeUndefined(); + }); +}); From 998dba0dd1a37023c91bc107767ce0381a0d300f Mon Sep 17 00:00:00 2001 From: WANG Qingmin <75425799+FriendlyPasser@users.noreply.github.com> Date: Mon, 14 Sep 2026 18:50:09 +0800 Subject: [PATCH 2/2] test(sessions): cover config parsing and invalid offer targets Signed-off-by: WANG Qingmin <75425799+FriendlyPasser@users.noreply.github.com> --- .../internal/handlers/sessions_test.go | 35 ++++++++++++++ .../pkg/types/session_turn_detection_test.go | 47 +++++++++++++++++++ 2 files changed, 82 insertions(+) diff --git a/control-plane/internal/handlers/sessions_test.go b/control-plane/internal/handlers/sessions_test.go index 551d41b71..58e7df588 100644 --- a/control-plane/internal/handlers/sessions_test.go +++ b/control-plane/internal/handlers/sessions_test.go @@ -529,3 +529,38 @@ func TestStartSessionRejectsTurnDetectionForOpenRouter(t *testing.T) { require.Equal(t, http.StatusBadRequest, rec.Code) require.Contains(t, rec.Body.String(), "turn_detection requires") } + +func TestSessionOfferRejectsInvalidRegisteredTargets(t *testing.T) { + gin.SetMode(gin.TestMode) + t.Setenv("OPENAI_API_KEY", "test-key") + original := http.DefaultClient.Transport + t.Cleanup(func() { http.DefaultClient.Transport = original }) + http.DefaultClient.Transport = roundTripFunc(func(req *http.Request) (*http.Response, error) { + t.Fatal("invalid target must not reach the provider") + return nil, nil + }) + for _, tc := range []struct { + name, target, provider, transport string + status int + message string + }{ + {"empty target", "", "openai", "webrtc", http.StatusBadRequest, "session target must be"}, + {"malformed target", "support", "openai", "webrtc", http.StatusBadRequest, "session target must be"}, + {"missing session", "support.missing", "openai", "webrtc", http.StatusNotFound, "session not registered"}, + {"wrong provider", "support.voice", "openrouter", "audio_turns", http.StatusBadRequest, "registered session must use"}, + {"wrong transport", "support.voice", "openai", "websocket", http.StatusBadRequest, "registered session must use"}, + } { + t.Run(tc.name, func(t *testing.T) { + agent := sessionTestAgent() + raw := agent.Metadata.Custom["sessions"].([]interface{})[0].(map[string]interface{}) + raw["provider"], raw["transport"] = tc.provider, tc.transport + router := gin.New() + router.POST("/:session_id/realtime-offer", SessionRealtimeOfferHandler(&nodeRESTStorageStub{agent: agent})) + rec := httptest.NewRecorder() + router.ServeHTTP(rec, httptest.NewRequest(http.MethodPost, + "/sess-1/realtime-offer?provider=openai&transport=webrtc&target="+url.QueryEscape(tc.target), strings.NewReader("v=0"))) + require.Equal(t, tc.status, rec.Code, rec.Body.String()) + require.Contains(t, rec.Body.String(), tc.message) + }) + } +} diff --git a/control-plane/pkg/types/session_turn_detection_test.go b/control-plane/pkg/types/session_turn_detection_test.go index 171dbc7be..14d05197e 100644 --- a/control-plane/pkg/types/session_turn_detection_test.go +++ b/control-plane/pkg/types/session_turn_detection_test.go @@ -3,6 +3,7 @@ package types import ( "encoding/json" "math" + "strings" "testing" ) @@ -75,3 +76,49 @@ func TestNormalizeTurnDetectionRejectsInvalidOptions(t *testing.T) { t.Fatalf("changed openrouter defaults: %+v, %v", config, err) } } + +func TestParseSessionTurnDetection(t *testing.T) { + for _, tc := range []struct { + name, provider, transport, raw, expected, wantError string + }{ + {name: "omitted", provider: "openai", transport: "webrtc", expected: `{"type":"server_vad","threshold":0.5,"prefix_padding_ms":300,"silence_duration_ms":500,"create_response":true,"interrupt_response":true}`}, + {name: "null", provider: "openai", transport: "webrtc", raw: " null ", expected: `{"type":"server_vad","threshold":0.5,"prefix_padding_ms":300,"silence_duration_ms":500,"create_response":true,"interrupt_response":true}`}, + {name: "server explicit zero and false", provider: "openai", transport: "webrtc", raw: `{"type":"server_vad","threshold":0,"prefix_padding_ms":0,"silence_duration_ms":0,"create_response":false,"interrupt_response":false}`, expected: `{"type":"server_vad","threshold":0,"prefix_padding_ms":0,"silence_duration_ms":0,"create_response":false,"interrupt_response":false}`}, + {name: "semantic", provider: "openai", transport: "websocket", raw: `{"type":"semantic_vad","eagerness":"low"}`, expected: `{"type":"semantic_vad","create_response":true,"interrupt_response":true,"eagerness":"low"}`}, + {name: "other provider omitted", provider: "openrouter", transport: "audio_turns", expected: `null`}, + {name: "other provider configured", provider: "openrouter", transport: "audio_turns", raw: `{"type":"server_vad"}`, wantError: "requires provider=openai"}, + {name: "malformed JSON", provider: "openai", transport: "webrtc", raw: `{`, wantError: "must be an object"}, + {name: "array", provider: "openai", transport: "webrtc", raw: `[]`, wantError: "must be an object"}, + {name: "missing type", provider: "openai", transport: "webrtc", raw: `{}`, wantError: "turn_detection.type"}, + {name: "unknown field", provider: "openai", transport: "webrtc", raw: `{"type":"server_vad","typo":true}`, wantError: "invalid turn_detection"}, + {name: "incorrect case", provider: "openai", transport: "webrtc", raw: `{"Type":"server_vad"}`, wantError: "unknown turn_detection field"}, + {name: "wrong boolean type", provider: "openai", transport: "webrtc", raw: `{"type":"server_vad","create_response":"false"}`, wantError: "invalid turn_detection"}, + {name: "null field", provider: "openai", transport: "webrtc", raw: `{"type":"server_vad","threshold":null}`, wantError: "threshold must not be null"}, + {name: "server with eagerness", provider: "openai", transport: "webrtc", raw: `{"type":"server_vad","eagerness":"low"}`, wantError: "eagerness is unsupported"}, + {name: "empty eagerness", provider: "openai", transport: "webrtc", raw: `{"type":"semantic_vad","eagerness":""}`, wantError: "eagerness must be"}, + {name: "mixed mode fields", provider: "openai", transport: "webrtc", raw: `{"type":"semantic_vad","threshold":0.5}`, wantError: "unsupported for semantic_vad"}, + } { + t.Run(tc.name, func(t *testing.T) { + config, err := ParseSessionTurnDetection(tc.provider, tc.transport, json.RawMessage(tc.raw)) + if tc.wantError != "" { + if err == nil || !strings.Contains(err.Error(), tc.wantError) { + t.Fatalf("expected error containing %q, got config=%+v err=%v", tc.wantError, config, err) + } + if config != nil { + t.Fatalf("invalid input returned usable config: %+v", config) + } + return + } + if err != nil { + t.Fatal(err) + } + body, err := json.Marshal(config) + if err != nil { + t.Fatal(err) + } + if string(body) != tc.expected { + t.Fatalf("got %s, want %s", body, tc.expected) + } + }) + } +}