fix(session): a reply spoken over ends its turn, not the socket; it writes on the session's context and leaves the next reply's floor alone

Claude-Session: https://claude.ai/code/session_01AGuBEqCjuman4E9FdKEgeV

LLM.md

@@ -33,6 +33,11 @@ Emitted: `session.created`, `input_audio_buffer.speech_started`/`.speech_stopped
 `response.created`, `response.audio_transcript.delta`, `response.audio.delta`
 (base64 PCM16 @ 24 kHz), `response.done`, `error`.
 
+A reply works on a context a barge-in cancels and writes the socket on the session's,
+which nothing cancels: coder/websocket closes the connection when a write's context is
+done, so a write on the cancelled one ended the conversation, not the turn. A reply
+clears the floor on its way out only while the floor is still its own (`finish`).
+
 ## The three seams
 
 `Turn` — audio in, floor out (`Quiet`/`Held`/`Yielded`). `Hush` ships with it and

session.go

@@ -19,9 +19,9 @@ import (
 type Floor int
 
 const (
-	Quiet  Floor = iota // nobody is speaking
-	Held                // the speaker has the floor and is still using it
-	Yielded             // the speaker just finished; a reply is due now
+	Quiet   Floor = iota // nobody is speaking
+	Held                 // the speaker has the floor and is still using it
+	Yielded              // the speaker just finished; a reply is due now
 )
 
 // Turn decides when a speaker has finished.
@@ -78,6 +78,7 @@ type Session struct {
 	// while audio is still queued.
 	mu     sync.Mutex
 	stop   context.CancelFunc
+	turn   int64        // which reply stop belongs to
 	item   atomic.Int64 // conversation item counter, for truncate
 	spoken atomic.Int64 // ms of audio sent for the item in flight
 	heard  atomic.Int64 // ms of audio taken in
@@ -212,6 +213,11 @@ func (s *Session) hush() {
 }
 
 // reply composes an answer and speaks it, a sentence at a time.
+//
+// The WORK runs on ctx, which a barge-in cancels; the SOCKET is written on parent,
+// which nothing cancels. coder/websocket closes the whole connection when a write's
+// context is done, so a reply that wrote on its own cancelled context after being
+// spoken over ended the conversation instead of the turn.
 func (s *Session) reply(parent context.Context) {
 	ctx, stop := context.WithCancel(parent)
 	s.mu.Lock()
@@ -221,24 +227,28 @@ func (s *Session) reply(parent context.Context) {
 		return
 	}
 	s.stop = stop
+	s.turn++
+	turn := s.turn
 	s.mu.Unlock()
-	defer s.hush()
+	defer s.finish(turn, stop)
 
 	item := fmt.Sprintf("%s-%d", s.ID, s.item.Add(1))
 	s.spoken.Store(0)
 
 	words, err := s.Mind.Reply(ctx)
 	if err != nil {
-		s.fail(ctx, err.Error())
+		if ctx.Err() == nil { // a turn that was spoken over is not an error
+			s.fail(parent, err.Error())
+		}
 		return
 	}
-	_ = s.send(ctx, rt.ServerEventBase{Type: rt.ServerEventTypeResponseCreated, EventID: item})
+	_ = s.send(parent, rt.ServerEventBase{Type: rt.ServerEventTypeResponseCreated, EventID: item})
 
 	for line := range Sentences(words, 240) {
 		if ctx.Err() != nil {
 			return // interrupted: stop synthesising what will not be heard
 		}
-		_ = s.send(ctx, transcript{
+		_ = s.send(parent, transcript{
 			ServerEventBase: rt.ServerEventBase{Type: rt.ServerEventTypeResponseAudioTranscriptDelta, EventID: item},
 			ItemID:          item,
 			Delta:           line + " ",
@@ -246,7 +256,7 @@ func (s *Session) reply(parent context.Context) {
 		pcm, err := s.Speech.Say(ctx, s.Model, s.Voice, line)
 		if err != nil {
 			if ctx.Err() == nil {
-				s.fail(ctx, "speech: "+err.Error())
+				s.fail(parent, "speech: "+err.Error())
 			}
 			return
 		}
@@ -256,7 +266,7 @@ func (s *Session) reply(parent context.Context) {
 		ms := int64(len(pcm)) * 1000 / (Spoken * 2)
 		s.spoken.Add(ms)
 		s.said.Add(ms)
-		if err := s.send(ctx, audio{
+		if err := s.send(parent, audio{
 			ServerEventBase: rt.ServerEventBase{Type: rt.ServerEventTypeResponseAudioDelta, EventID: item},
 			ItemID:          item,
 			Delta:           base64.StdEncoding.EncodeToString(pcm),
@@ -264,7 +274,19 @@ func (s *Session) reply(parent context.Context) {
 			return
 		}
 	}
-	_ = s.send(ctx, rt.ServerEventBase{Type: rt.ServerEventTypeResponseDone, EventID: item})
+	_ = s.send(parent, rt.ServerEventBase{Type: rt.ServerEventTypeResponseDone, EventID: item})
+}
+
+// finish ends a reply. It clears the floor only while the floor is still this
+// reply's: one spoken over has already given it up, and the reply that took it
+// next must not be cancelled by the old one leaving.
+func (s *Session) finish(turn int64, stop context.CancelFunc) {
+	s.mu.Lock()
+	if s.turn == turn {
+		s.stop = nil
+	}
+	s.mu.Unlock()
+	stop()
 }
 
 // audio and transcript carry the two deltas the protocol defines but the

session_test.go

@@ -0,0 +1,124 @@
+package voice
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"io"
+	"log/slog"
+	"net"
+	"sync"
+	"testing"
+	"time"
+)
+
+// tape is a socket that behaves as coder/websocket does on the one point that
+// matters here: a write whose context is already done closes the connection.
+type tape struct {
+	mu     sync.Mutex
+	sent   []string
+	closed bool
+}
+
+func (t *tape) Read(ctx context.Context) ([]byte, error) {
+	<-ctx.Done()
+	return nil, ctx.Err()
+}
+
+func (t *tape) Write(ctx context.Context, msg []byte) error {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+	if t.closed {
+		return net.ErrClosed
+	}
+	if ctx.Err() != nil {
+		t.closed = true
+		return ctx.Err()
+	}
+	var head struct{ Type string }
+	_ = json.Unmarshal(msg, &head)
+	t.sent = append(t.sent, head.Type)
+	return nil
+}
+
+func (t *tape) Close() error { return nil }
+
+func (t *tape) open() bool {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+	return !t.closed
+}
+
+// thinking is a mind that has begun to answer and has said nothing yet, the
+// moment a speaker most often talks over. It answers, or fails, only once told
+// to stop.
+type thinking struct {
+	started chan struct{}
+	fails   bool
+}
+
+func (m *thinking) Hear(context.Context, []byte) {}
+func (m *thinking) See([]byte)                   {}
+func (m *thinking) Reply(ctx context.Context) (<-chan string, error) {
+	close(m.started)
+	if m.fails {
+		<-ctx.Done()
+		return nil, errors.New("nothing was said")
+	}
+	out := make(chan string)
+	go func() {
+		defer close(out)
+		<-ctx.Done()
+	}()
+	return out, nil
+}
+
+// TestSpeakingOverAReplyEndsTheTurnNotTheConversation: barge-in cancels the reply,
+// and whatever the cancelled reply does on its way out must leave the socket open
+// for the next turn.
+func TestSpeakingOverAReplyEndsTheTurnNotTheConversation(t *testing.T) {
+	for _, fails := range []bool{false, true} {
+		sock := &tape{}
+		mind := &thinking{started: make(chan struct{}), fails: fails}
+		s := &Session{ID: "s", Sock: sock, Mind: mind, Turn: &Hush{}, Log: slog.New(slog.NewTextHandler(io.Discard, nil))}
+
+		done := make(chan struct{})
+		go func() {
+			s.reply(context.Background())
+			close(done)
+		}()
+		<-mind.started
+		s.hush() // the speaker talks over it
+		select {
+		case <-done:
+		case <-time.After(5 * time.Second):
+			t.Fatal("a spoken-over reply never finished")
+		}
+		if !sock.open() {
+			t.Fatalf("mind fails=%v: the cancelled reply closed the socket; sent %v", fails, sock.sent)
+		}
+		for _, sent := range sock.sent {
+			if sent == "error" {
+				t.Fatalf("mind fails=%v: a turn that was spoken over was reported as an error", fails)
+			}
+		}
+	}
+}
+
+// TestALeavingReplyDoesNotCancelTheNextOne: the reply that was spoken over
+// finishes after the next has taken the floor, and must leave it alone.
+func TestALeavingReplyDoesNotCancelTheNextOne(t *testing.T) {
+	s := &Session{}
+	stop1, stop2 := func() {}, func() {}
+	s.stop, s.turn = stop1, 1
+	s.hush()                  // spoken over
+	s.stop, s.turn = stop2, 2 // the next reply took the floor
+	s.finish(1, func() {})
+	if s.stop == nil {
+		t.Fatal("the old reply leaving cleared the new reply's floor")
+	}
+	s.finish(2, func() {})
+	if s.stop != nil {
+		t.Fatal("a reply leaving did not give up its own floor")
+	}
+}