fix(realtime): honor interrupt_response

Keep active responses running when VAD interruption is disabled. Skip
an overlapping automatic response while preserving its transcript and
conversation history. Default interruption and explicit client actions
keep their existing behavior.

Add regression coverage for both VAD modes and document skipped turns.

Assisted-by: Codex:gpt-6
(cherry picked from commit f54023a2f9)
Signed-off-by: Ettore Di Giacinto <mudler@localai.io>

core/http/endpoints/openai/realtime.go

@@ -285,7 +285,9 @@ func (session *Session) issueCommit(parent context.Context, source respcoord.Sou
 	session.commitOrderMu.Lock()
 	defer session.commitOrderMu.Unlock()
 	slot := session.nextCommitSlotLocked()
-	session.respSink.issue(parent, source, func(ctx context.Context) {
+	// A non-interrupting VAD commit must not supersede a response at speech end.
+	ifIdle := source == respcoord.SourceVAD && !session.interruptResponseEnabled()
+	session.respSink.issueWithPolicy(parent, source, ifIdle, func(ctx context.Context) {
 		run(ctx, slot)
 	})
 }

core/http/endpoints/openai/realtime_interrupt_test.go

@@ -0,0 +1,135 @@
+// SPDX-License-Identifier: MIT
+package openai
+
+import (
+	"context"
+	"encoding/json"
+	"time"
+
+	"github.com/mudler/LocalAI/core/config"
+	"github.com/mudler/LocalAI/core/http/endpoints/openai/respcoord"
+	"github.com/mudler/LocalAI/core/http/endpoints/openai/types"
+	"github.com/mudler/LocalAI/core/schema"
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+)
+
+var _ = Describe("realtime interrupt_response", func() {
+	decode := func(mode, setting string) *types.TurnDetectionUnion {
+		var td types.TurnDetectionUnion
+		Expect(json.Unmarshal([]byte(`{"type":"`+mode+`"`+setting+`}`), &td)).To(Succeed())
+		return &td
+	}
+	holdResponse := func(s *Session) context.Context {
+		started := make(chan context.Context, 1)
+		s.respSink.issue(context.Background(), respcoord.SourceClient, func(ctx context.Context) {
+			started <- ctx
+			<-ctx.Done()
+		})
+		DeferCleanup(s.respSink.shutdown)
+		var ctx context.Context
+		Eventually(started).Should(Receive(&ctx))
+		return ctx
+	}
+
+	DescribeTable("controls cancellation at speech onset",
+		func(mode, setting string, interrupts bool) {
+			s := &Session{TurnDetection: decode(mode, setting), InputSampleRate: 16000,
+				InputAudioBuffer: make([]byte, 32000), ModelConfig: &config.ModelConfig{},
+				ModelInterface: &fakeModel{vadSegments: []schema.VADSegment{{Start: 0.1}}}, respSink: newResponseSink()}
+			ctx := holdResponse(s)
+			tr := &fakeTransport{}
+			sink := newTurnSink(s, &Conversation{}, tr, newLiveTurnState(s, tr), context.Background(), time.Now())
+			vadTick(sink, 0.5)
+			Expect(tr.countEvents(types.ServerEventTypeInputAudioBufferSpeechStarted)).To(Equal(1))
+			Expect(ctx.Err() != nil).To(Equal(interrupts))
+		},
+		Entry("server false", "server_vad", `,"interrupt_response":false`, false),
+		Entry("semantic false", "semantic_vad", `,"interrupt_response":false`, false),
+		Entry("server true", "server_vad", `,"interrupt_response":true`, true),
+		Entry("semantic true", "semantic_vad", `,"interrupt_response":true`, true),
+		Entry("server omitted", "server_vad", "", true),
+		Entry("semantic omitted", "semantic_vad", "", true),
+	)
+
+	DescribeTable("preserves the active response and commits overlapping speech",
+		func(mode string) {
+			s := &Session{TurnDetection: decode(mode, `,"interrupt_response":false`),
+				InputAudioTranscription: &types.AudioTranscription{},
+				ModelInterface:          &fakeModel{cfg: &config.ModelConfig{}, transcribeFinal: &schema.TranscriptionResult{Text: "next turn"}},
+				ModelConfig:             &config.ModelConfig{}, respSink: newResponseSink()}
+			ctx := holdResponse(s)
+			conv, tr := &Conversation{}, &fakeTransport{}
+			committed := make(chan struct{})
+			s.issueCommit(context.Background(), respcoord.SourceVAD, func(turnCtx context.Context, slot *commitSlot) {
+				commitUtteranceWithTranscript(turnCtx, []byte{1, 2, 3, 4}, nil, nil, "", s, conv, tr, slot)
+				close(committed)
+			})
+			Eventually(committed).Should(BeClosed())
+			Expect(ctx.Err()).NotTo(HaveOccurred())
+			Expect(tr.countEvents(types.ServerEventTypeResponseCreated)).To(BeZero())
+			Expect(tr.countEvents(types.ServerEventTypeConversationItemInputAudioTranscriptionCompleted)).To(Equal(1))
+			conv.Lock.Lock()
+			defer conv.Lock.Unlock()
+			Expect(conv.Items).To(HaveLen(1))
+			Expect(conv.Items[0].User.Content[0].Transcript).To(Equal("next turn"))
+		},
+		Entry("server", "server_vad"), Entry("semantic", "semantic_vad"),
+	)
+
+	It("applies session updates to interruption and echoes false", func() {
+		s := &Session{TurnDetection: defaultTurnDetection(nil), ModelConfig: &config.ModelConfig{}}
+		var update types.SessionUpdateEvent
+		Expect(json.Unmarshal([]byte(`{"type":"session.update","session":{"type":"realtime","audio":{"input":{"turn_detection":{"type":"semantic_vad","interrupt_response":false}}}}}`), &update)).To(Succeed())
+		Expect(updateSession(s, &update.Session, nil, nil, nil, nil, nil, nil)).To(Succeed())
+		Expect(s.interruptResponseEnabled()).To(BeFalse())
+		raw, err := json.Marshal(s.ToServer())
+		Expect(err).NotTo(HaveOccurred())
+		Expect(string(raw)).To(ContainSubstring(`"interrupt_response":false`))
+		Expect(json.Unmarshal([]byte(`{"type":"session.update","session":{"type":"realtime","instructions":"Keep talking"}}`), &update)).To(Succeed())
+		Expect(updateSession(s, &update.Session, nil, nil, nil, nil, nil, nil)).To(Succeed())
+		Expect(s.interruptResponseEnabled()).To(BeFalse())
+		s.TurnDetection = decode("server_vad", `,"interrupt_response":true`)
+		Expect(s.interruptResponseEnabled()).To(BeTrue())
+	})
+
+	DescribeTable("starts commits when idle and retains explicit client cancellation",
+		func(source respcoord.Source) {
+			s := &Session{TurnDetection: decode("server_vad", `,"interrupt_response":false`), respSink: newResponseSink()}
+			DeferCleanup(s.respSink.shutdown)
+			started := make(chan context.Context, 1)
+			s.issueCommit(context.Background(), source, func(ctx context.Context, slot *commitSlot) {
+				defer close(slot.done)
+				started <- ctx
+				<-ctx.Done()
+			})
+			var ctx context.Context
+			Eventually(started).Should(Receive(&ctx))
+			Expect(ctx.Err()).NotTo(HaveOccurred())
+			s.respSink.cancel(respcoord.SourceClient)
+			Expect(ctx.Err()).To(MatchError(context.Canceled))
+		}, Entry("VAD", respcoord.SourceVAD), Entry("client", respcoord.SourceClient),
+	)
+
+	It("keeps explicit client supersession when interruption is disabled", func() {
+		s := &Session{TurnDetection: decode("server_vad", `,"interrupt_response":false`), respSink: newResponseSink()}
+		ctx := holdResponse(s)
+		done := make(chan struct{})
+		s.issueCommit(context.Background(), respcoord.SourceClient, func(_ context.Context, slot *commitSlot) {
+			close(slot.done)
+			close(done)
+		})
+		Eventually(done).Should(BeClosed())
+		Expect(ctx.Err()).To(MatchError(context.Canceled))
+	})
+
+	DescribeTable("echoes an explicit false through JSON",
+		func(mode string) {
+			raw, err := json.Marshal(decode(mode, `,"interrupt_response":false`))
+			Expect(err).NotTo(HaveOccurred())
+			var fields map[string]any
+			Expect(json.Unmarshal(raw, &fields)).To(Succeed())
+			Expect(fields).To(HaveKeyWithValue("interrupt_response", false))
+		}, Entry("server", "server_vad"), Entry("semantic", "semantic_vad"),
+	)
+})

core/http/endpoints/openai/realtime_respcoord.go

@@ -58,11 +58,17 @@ func newResponseSink() *responseSink {
 // race. Non-blocking: the superseded response drains concurrently and its later
 // Finished is ignored as stale.
 func (s *responseSink) issue(parent context.Context, source respcoord.Source, run func(ctx context.Context)) {
+	s.issueWithPolicy(parent, source, false, run)
+}
+
+// issueWithPolicy checks IfIdle inside the coordinator's serialized transition,
+// avoiding a race between checking for an active response and starting one.
+func (s *responseSink) issueWithPolicy(parent context.Context, source respcoord.Source, ifIdle bool, run func(ctx context.Context)) {
 	id := respcoord.ResponseID(s.seq.Add(1))
 	s.mu.Lock()
 	s.bodies[id] = responseBody{parent: parent, run: run}
 	s.mu.Unlock()
-	if err := s.coord.Apply(respcoord.Start{ID: id, Source: source}); err != nil {
+	if err := s.coord.Apply(respcoord.Start{ID: id, Source: source, IfIdle: ifIdle}); err != nil {
 		xlog.Error("respcoord: start failed", "error", err)
 	}
 }
@@ -102,33 +108,9 @@ func (s *responseSink) shutdown() {
 func (s *responseSink) Perform(e respcoord.Effect) {
 	switch eff := e.(type) {
 	case respcoord.StartResponse:
-		s.mu.Lock()
-		body := s.bodies[eff.ID]
-		delete(s.bodies, eff.ID)
-		parent := body.parent
-		if parent == nil {
-			parent = context.Background()
-		}
-		ctx, cancel := context.WithCancel(parent)
-		s.cancels[eff.ID] = cancel
-		s.mu.Unlock()
-
-		s.wg.Go(func() {
-			defer func() {
-				s.mu.Lock()
-				delete(s.cancels, eff.ID)
-				s.mu.Unlock()
-				// Report completion. If this response was superseded/cancelled
-				// the id is stale and the coordinator ignores it (so the
-				// terminal is never emitted twice).
-				if err := s.coord.Apply(respcoord.Finished{ID: eff.ID}); err != nil {
-					xlog.Error("respcoord: finished apply failed", "error", err)
-				}
-			}()
-			if body.run != nil {
-				body.run(ctx)
-			}
-		})
+		s.runBody(eff.ID, false)
+	case respcoord.SkipResponse:
+		s.runBody(eff.ID, true)
 	case respcoord.CancelResponse:
 		s.mu.Lock()
 		cancel := s.cancels[eff.ID]
@@ -141,3 +123,43 @@ func (s *responseSink) Perform(e respcoord.Effect) {
 		// Wiring the authoritative single terminal here is the next step.
 	}
 }
+
+// runBody runs under the coordinator lock, just like Perform.
+func (s *responseSink) runBody(id respcoord.ResponseID, skipped bool) {
+	s.mu.Lock()
+	body := s.bodies[id]
+	delete(s.bodies, id)
+	parent := body.parent
+	if parent == nil {
+		parent = context.Background()
+	}
+	ctx, cancel := context.WithCancel(parent)
+	if skipped {
+		// Commit bodies use the session context for transcription and this cancelled
+		// response context to append the user item without generating a response.
+		cancel()
+	} else {
+		s.cancels[id] = cancel
+	}
+	s.mu.Unlock()
+
+	s.wg.Go(func() {
+		defer cancel()
+		defer func() {
+			s.mu.Lock()
+			delete(s.cancels, id)
+			s.mu.Unlock()
+			// Report completion. If this response was superseded/cancelled
+			// the id is stale and the coordinator ignores it (so the
+			// terminal is never emitted twice).
+			if !skipped {
+				if err := s.coord.Apply(respcoord.Finished{ID: id}); err != nil {
+					xlog.Error("respcoord: finished apply failed", "error", err)
+				}
+			}
+		}()
+		if body.run != nil {
+			body.run(ctx)
+		}
+	})
+}

core/http/endpoints/openai/realtime_semantic_vad.go

@@ -389,3 +389,19 @@ func defaultTurnDetection(cfg *config.ModelConfig) *types.TurnDetectionUnion {
 		},
 	}
 }
+
+// interruptResponseEnabled treats omission as true while preserving an explicit
+// false. The session lock also makes updates visible to the VAD goroutine.
+func (s *Session) interruptResponseEnabled() bool {
+	sessionLock.Lock()
+	defer sessionLock.Unlock()
+	var enabled *bool
+	if td := s.TurnDetection; td != nil {
+		if td.ServerVad != nil {
+			enabled = td.ServerVad.InterruptResponse
+		} else if td.SemanticVad != nil {
+			enabled = td.SemanticVad.InterruptResponse
+		}
+	}
+	return enabled == nil || *enabled
+}

core/http/endpoints/openai/realtime_turncoord.go

@@ -92,7 +92,9 @@ func (s *turnSink) Perform(e turncoord.Effect) {
 			s.lts.feedNewAudio(s.onsetAudio)
 		}
 	case turncoord.BargeIn:
-		s.session.respSink.cancel(respcoord.SourceVAD)
+		if s.session.interruptResponseEnabled() {
+			s.session.respSink.cancel(respcoord.SourceVAD)
+		}
 	case turncoord.EmitSpeechStarted:
 		sendEvent(s.transport, types.InputAudioBufferSpeechStartedEvent{
 			ServerEventBase: types.ServerEventBase{EventID: "event_TODO"},

core/http/endpoints/openai/respcoord/respcoord.go

@@ -122,6 +122,8 @@ type Event interface {
 type Start struct {
 	ID     ResponseID
 	Source Source
+	// IfIdle skips this attempt when a response is already active.
+	IfIdle bool
 }
 
 // Finished reports that the response goroutine for ID reached its own terminal.
@@ -152,7 +154,7 @@ func (Shutdown) String() string   { return "Shutdown" }
 // Returning effects as data (rather than firing callbacks inside the
 // transition) keeps Next pure and exhaustively testable, and lets the
 // Coordinator decide how/when to perform them. Exhaustively:
-// CancelResponse | StartResponse | EmitTerminal.
+// CancelResponse | StartResponse | SkipResponse | EmitTerminal.
 type Effect interface {
 	isEffect()
 	String() string
@@ -164,6 +166,10 @@ type CancelResponse struct{ ID ResponseID }
 // StartResponse: spawn the response goroutine for ID.
 type StartResponse struct{ ID ResponseID }
 
+// SkipResponse keeps the active response. The sink must release the skipped
+// attempt's resources; a committed audio turn can still retain its transcript.
+type SkipResponse struct{ ID ResponseID }
+
 // EmitTerminal: send response.done for ID with Status.
 type EmitTerminal struct {
 	ID     ResponseID
@@ -172,10 +178,12 @@ type EmitTerminal struct {
 
 func (CancelResponse) isEffect() {}
 func (StartResponse) isEffect()  {}
+func (SkipResponse) isEffect()   {}
 func (EmitTerminal) isEffect()   {}
 
 func (e CancelResponse) String() string { return fmt.Sprintf("CancelResponse(%d)", e.ID) }
 func (e StartResponse) String() string  { return fmt.Sprintf("StartResponse(%d)", e.ID) }
+func (e SkipResponse) String() string   { return fmt.Sprintf("SkipResponse(%d)", e.ID) }
 func (e EmitTerminal) String() string {
 	return fmt.Sprintf("EmitTerminal(%d,%s)", e.ID, e.Status)
 }
@@ -213,6 +221,9 @@ func Next(s State, e Event) (State, []Effect, error) {
 	case Active:
 		switch ev := e.(type) {
 		case Start:
+			if ev.IfIdle {
+				return st, []Effect{SkipResponse{ID: ev.ID}}, nil
+			}
 			return Active{ID: ev.ID}, []Effect{
 				CancelResponse{ID: st.ID},
 				EmitTerminal{ID: st.ID, Status: StatusCancelled},

core/http/endpoints/openai/respcoord/respcoord_test.go

@@ -85,6 +85,12 @@ var _ = Describe("respcoord.Next", func() {
 		Entry("idle+start -> active, spawns response",
 			Idle{}, Start{ID: 1, Source: SourceClient},
 			Active{ID: 1}, []Effect{StartResponse{ID: 1}}),
+		Entry("idle+conditional start -> active",
+			Idle{}, Start{ID: 2, Source: SourceVAD, IfIdle: true},
+			Active{ID: 2}, []Effect{StartResponse{ID: 2}}),
+		Entry("active+conditional start -> retain active and skip new response",
+			Active{ID: 1}, Start{ID: 2, Source: SourceVAD, IfIdle: true},
+			Active{ID: 1}, []Effect{SkipResponse{ID: 2}}),
 		Entry("idle+cancel -> idle, no-op",
 			Idle{}, Cancel{Source: SourceVAD},
 			Idle{}, []Effect(nil)),

core/http/endpoints/openai/types/types.go

@@ -602,7 +602,7 @@ type ServerVad struct {
 	CreateResponse bool `json:"create_response,omitempty"`
 
 	// Whether or not to automatically interrupt any ongoing response with output to the default conversation (i.e. conversation of auto) when a VAD start event occurs.
-	InterruptResponse bool `json:"interrupt_response,omitempty"`
+	InterruptResponse *bool `json:"interrupt_response,omitempty"`
 
 	// Used only for server_vad mode. Amount of audio to include before the VAD detected speech (in milliseconds). Defaults to 300ms.
 	PrefixPaddingMs int64 `json:"prefix_padding_ms,omitempty"`
@@ -636,7 +636,7 @@ type RealtimeSessionSemanticVad struct {
 	CreateResponse bool `json:"create_response,omitempty"`
 
 	// Whether or not to automatically interrupt any ongoing response with output to the default conversation (i.e. conversation of auto) when a VAD start event occurs.
-	InterruptResponse bool `json:"interrupt_response,omitempty"`
+	InterruptResponse *bool `json:"interrupt_response,omitempty"`
 
 	// Used only for semantic_vad mode. The eagerness of the model to respond. low will wait longer for the user to continue speaking, high will respond more quickly. auto is the default and is equivalent to medium. low, medium, and high have max timeouts of 8s, 4s, and 2s respectively.
 	Eagerness string `json:"eagerness,omitempty"`

docs/content/features/openai-realtime.md

@@ -91,6 +91,40 @@ curl -X POST http://localhost:8080/backend/load \
 
 The endpoint is not realtime-specific - it pre-loads any model. See [Backend Monitor]({{%relref "operations/backend-monitor" %}}) for the full request/response reference (it is the inverse of `/backend/shutdown`).
 
+### Controlling response interruption
+
+By default, speech detected by either VAD mode interrupts the active response.
+Set `interrupt_response: false` to let the response finish while the user speaks:
+
+```json
+{
+  "type": "session.update",
+  "session": {
+    "type": "realtime",
+    "audio": {
+      "input": {
+        "turn_detection": {
+          "type": "server_vad",
+          "silence_duration_ms": 500,
+          "interrupt_response": false
+        }
+      }
+    }
+  }
+}
+```
+
+This option also applies to `semantic_vad`. Speech events and transcription
+continue. If a response is active when another turn commits, LocalAI retains
+that turn in conversation history but skips its automatic response. It does
+not queue a response for that turn. A later turn can generate a response once
+the active one finishes. To request a response manually, wait for both
+`response.done` and the overlapping turn's `conversation.item.added` event, then
+send `response.create`. Transcription can finish after the active response.
+Explicit `response.cancel` and client response creation keep their existing
+behavior. When replacing the turn-detection settings, omit `interrupt_response`
+or set it to `true` to enable interruption again.
+
 ### Turn detection
 
 Turn detection decides when the user has finished speaking and the pipeline should respond. Two modes are supported, matching the OpenAI session schema: