answer the room in the gap the floor leaves

The library had both directions and no service: attend.go ended in an empty
main. This closes the loop — a turn is transcribed, our model answers it, and
hanzoai/speech gives that answer a voice the room hears.

The floor decides when. It frees itself `rest` after the holder stops, which is
the pause a person waits out before speaking, and it fills again within `notice`
of anyone starting — so an answer still playing when somebody talks is cut off
there and the track comes down. Nothing here holds a second notion of turns.

Say takes a context and the bytes rather than a path: it returns when the room
has heard the audio or the moment it is cancelled, and unpublishes either way. A
track left up is a microphone nobody takes down, and every answer added another.

A session is now addressed where the open says it lives. A growing transcript is
a window in one process's memory and speech runs two replicas behind a Service
that round-robins, so pushing to the Service address is a 404 for about half of
every turn.

The attendant holds no LiveKit signing key: it asks meet for one room's token
the way a browser does, carrying one IAM identity to meet, the model and the
voice alike.

Co-authored-by: Hanzo Dev <[email protected]>

.hanzo/workflows/cicd.yml

@@ -0,0 +1,19 @@
+# The canonical caller. All real config lives in /hanzo.yml, which
+# platform.hanzo.ai reads too.
+#
+# It does NOT roll out: the attendant is pinned to an explicit tag in
+# hanzo/universe (charts/app/values/hanzo/attendant.yaml), rolled by a reviewed
+# change there. That is what hanzo.yml means by having no `deploy:` section — a
+# patch from CI would be restored from the universe pin on cd.hanzo.ai's next
+# poll anyway.
+name: CI/CD
+on:
+  push:
+    branches: [main]
+    tags: ['v*']
+  pull_request:
+  workflow_dispatch:
+jobs:
+  cicd:
+    uses: hanzoai/ci/.hanzo/workflows/build.yml@v1
+    secrets: inherit

Dockerfile

@@ -0,0 +1,25 @@
+# syntax=docker/dockerfile:1
+#
+# The attendant is a member of a room, not a service anything calls: it opens no
+# port and serves nothing. So the image is the binary and a CA bundle — the
+# rooms, the model and the voice are all reached over TLS, and that is the whole
+# of what it needs from a filesystem.
+#
+# Pure Go, no cgo: the Opus decoder is pion's, written in Go, so nothing here
+# links against libopus and the binary runs on a scratch-thin base.
+FROM golang:1.26.5-alpine AS build
+WORKDIR /src
+ENV GOPROXY=https://proxy.golang.org,direct
+COPY go.mod go.sum ./
+RUN --mount=type=cache,id=attendant-gomod,target=/go/pkg/mod go mod download
+COPY . .
+RUN --mount=type=cache,id=attendant-gomod,target=/go/pkg/mod \
+    --mount=type=cache,target=/root/.cache/go-build \
+    CGO_ENABLED=0 go build -trimpath -ldflags="-s -w" -o /attendant .
+
+FROM alpine:3.21
+RUN apk add --no-cache ca-certificates \
+    && addgroup -S hanzo && adduser -S hanzo -G hanzo
+COPY --from=build /attendant /usr/local/bin/attendant
+USER hanzo
+ENTRYPOINT ["/usr/local/bin/attendant"]

README.md

@@ -1,22 +1,30 @@
 # attendant
 
-A server-side participant for a LiveKit room. It joins, subscribes to the audio
-every human publishes, and transcribes **whoever holds the floor**.
+A participant in a LiveKit room that is not a person. It joins, subscribes to
+the audio every human publishes, transcribes **whoever holds the floor**, and
+answers them out loud.
 
 A meeting has one floor. At any instant one person is talking and the rest are
 listening, so a ten-person meeting does not need ten transcripts — it needs one,
 handed from speaker to speaker. That is the whole idea, and it is what makes the
-cost of transcribing a meeting independent of how many people are in it.
+cost of transcribing a meeting independent of how many people are in it. It is
+also what tells the attendant when it may speak: the floor is empty.
 
-    attendant ──► Floor ──► Scribe ──► hanzoai/speech
-     (a room)    (who)      (a turn)   (the words)
+    hears     attendant ──► Floor ──► Scribe ──► hanzoai/speech
+              (a room)      (who)     (a turn)   (the words)
+
+    answers   api.hanzo.ai ──► hanzoai/speech ──► Say
+              (what to say)    (the voice)        (into the gap)
 
 - `attend.go` — one room membership. Subscribes, decodes Opus to the 16 kHz mono
-  the speech service takes, and reports the room's active speakers.
+  the speech service takes, reports the room's active speakers, and publishes an
+  answer back into the room.
 - `floor.go` — who gets transcribed. No LiveKit in it: the room hands it a
   ranking of identities and frames of audio, and it decides.
 - `scribe.go` — turns into transcripts, over hanzoai/speech's growing-transcript
   API. One session at a time, and one settling at a time.
+- `answer.go` — what the attendant says back, and when.
+- `meet.go` — the way in, and the process.
 
 ## What the floor has to get right
 
@@ -56,6 +64,34 @@ number is the loss and not the overlap. An interjection that matters becomes the
 floor and its onset comes out of `lead`, so real turn-taking survives and only
 backchannel is dropped. A second stream would put the headcount back in the bill.
 
+## What answering has to get right
+
+**A turn ending is not an invitation; an empty floor is.** When a turn is
+transcribed the attendant has something it could say, and often somebody else is
+already saying something. So it waits for the floor to be EMPTY — the floor
+frees itself `rest` after the holder stops, which is the pause a person waits
+out before speaking. Nothing here has a second opinion about turns.
+
+**Being interrupted costs a syllable.** The floor fills within `notice` of
+anyone starting, and the attendant is not on it — it is dropped from the room's
+ranking and subscribes to nothing of its own — so the first person to make a
+sound takes the floor from an empty one, and that cancels the answer mid-word.
+The track comes down with it.
+
+**Nobody answers themselves.** A turn that ends while the attendant is talking
+asks for an answer that, by the time it is read, has already been given. The
+attendant answers only turns it has not answered, or it would reply to its own
+voice and then to that.
+
+**Two turns while thinking are one answer.** An answer is formed from the record
+at the moment it is formed, so turns that land in the meantime are IN it. The
+ask is a single slot: one is as good as three.
+
+**Saying nothing is an answer.** People in a meeting talk to each other as well
+as to the attendant, and the model is told to answer with nothing when the last
+thing said was not addressed to it. That is the only way it has of declining a
+floor it was not offered.
+
 ## Measured
 
 A meeting of three, then the same meeting with nine more people who came to
@@ -69,11 +105,37 @@ listen with their mics open (`TestTheFloorHoldsOneSession`):
 Four times the room, the same bill — **11.6× less audio decoded** than a
 transcript per person, and the nine who never spoke never opened one.
 
+## Running one
+
+The attendant holds no LiveKit signing key. It asks cloud's meet app for one
+room's token exactly as a browser does, so what it carries is one room's
+capability — and meet admits a **person**, so the attendant's identity is an IAM
+account like anyone else's, a member of the workspace whose room it sits in.
+Rooms are named the way meet names them: `<workspace>_<room>_<id>`.
+
+| | |
+|---|---|
+| `ROOM` | the room to join. Required. |
+| `HANZO_TOKEN` | its IAM identity, carried to meet, the model and the voice alike. Required. |
+| `MEET` | `https://api.hanzo.ai` |
+| `AI` | `https://api.hanzo.ai` |
+| `MODEL` | `best` |
+| `SPEECH` | `http://speech.hanzo.svc` |
+| `VOICE` | `af_heart` |
+| `NAME` | `Attendant`, the label the room shows |
+| `PROMPT` | what the model is told it is |
+
+It opens no port and answers nothing: it is a member of a room, and when the
+membership ends the process ends with it.
+
 ## Tests
 
-    go test -race ./...          # the floor, on a clock the test drives
+    go test -race ./...          # the floor and the answer, on a clock the test drives
     ./mutate.sh                  # breaks each fix, requires its test to go red
 
+`mutate.sh` rewrites source in place and restores it, so it is run before
+pushing and never on a CI runner that builds an image from the same tree.
+
 Live, with credentials in the environment:
 
     LK_URL / LK_KEY / LK_SECRET  # the real SFU: TestTheRoomAwardsTheFloor

answer.go

@@ -0,0 +1,243 @@
+// What the attendant says back.
+//
+// The floor already knows who is talking, so nothing here holds a second
+// opinion about turns. A turn ends, the attendant reads the meeting as it
+// stands, our model answers it, hanzoai/speech gives that answer a voice, and
+// the room hears it — but only while the floor is EMPTY. The floor frees itself
+// `rest` after the holder stops, which is the pause a person waits out before
+// speaking, and it fills again within `notice` of anyone starting: so an answer
+// still playing when somebody talks is cut off there. Interrupting the
+// attendant costs a syllable, not a paragraph.
+//
+// One answer at a time, always formed from the record at the moment it is
+// formed. Turns that end while the attendant is thinking do not each queue an
+// answer — they are in the record the one answer reads, which is what a person
+// catching up would do. And the attendant answers only what it has not answered
+// already, so its own voice is never something it replies to.
+package main
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"log"
+	"net/http"
+	"strings"
+	"time"
+)
+
+const (
+	// How long one answer may take, from asking the model to the last of the
+	// audio. Everything in an answer is a call to something else, and a call
+	// that never returns is an attendant that never speaks again; this is what
+	// makes that a lost answer rather than a wedged process.
+	patience = 2 * time.Minute
+
+	// The most a single answer may be. It is going to be READ ALOUD, and the
+	// room cannot skim it: this is about a paragraph, past which nobody is
+	// listening anyway and the floor should have gone back long ago.
+	most = 160
+)
+
+// Answer keeps the meeting's record like any scribe, and answers what lands in
+// it. It IS the scribe the floor is given, so a turn reaches it only once the
+// service has settled the words — there is nothing to reply to before that.
+type Answer struct {
+	*Notes
+
+	ai     string // our model, at api.hanzo.ai
+	model  string
+	speech string // hanzoai/speech, for the voice
+	voice  string
+	prompt string
+	who    string // the attendant's own name in the record
+	key    string // one identity, from IAM, carried to both
+
+	// A turn landing while an answer is being formed does not ask for a second
+	// answer: it is in the record that answer reads. One nudge is as good as
+	// three, so this holds one and a full one is dropped.
+	wake chan struct{}
+
+	// Turns already answered. Owned by the answering loop — one goroutine, so
+	// no lock, and no second answerer to share it with.
+	answered int
+}
+
+// Turn records the turn as the notes do, and says so when it ends. The nudge is
+// on Close and not on Turn because until Close there is no text: a turn is
+// answerable only once the service has settled it.
+func (a *Answer) Turn(speaker string, at time.Time) (Turn, error) {
+	t, err := a.Notes.Turn(speaker, at)
+	if err != nil {
+		return nil, err
+	}
+	return &landed{Turn: t, then: a.nudge}, nil
+}
+
+// landed is a turn that says so when it ends.
+type landed struct {
+	Turn
+	then func()
+}
+
+func (l *landed) Close() error {
+	err := l.Turn.Close()
+	l.then()
+	return err
+}
+
+func (a *Answer) nudge() {
+	select {
+	case a.wake <- struct{}{}:
+	default:
+	}
+}
+
+// tend answers until ctx is cancelled: the whole of the attendant's side of the
+// conversation. A failed answer is one answer lost and reported, never the end
+// of the meeting — the next turn asks again.
+func (a *Answer) tend(ctx context.Context, f *Floor, say func(context.Context, io.ReadCloser) error) {
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-a.wake:
+		}
+		if err := a.answer(ctx, f, say); err != nil && ctx.Err() == nil {
+			log.Printf("answer: %v", err)
+		}
+	}
+}
+
+// answer forms one answer and says it.
+func (a *Answer) answer(ctx context.Context, f *Floor, say func(context.Context, io.ReadCloser) error) error {
+	ctx, done := context.WithTimeout(ctx, patience)
+	defer done()
+
+	// Nobody speaks over the person holding the floor. Waiting here rather than
+	// at the end also means the model reads a meeting that has stopped moving.
+	if err := until(ctx, f, free); err != nil {
+		return err
+	}
+
+	// Only what has not been answered. A nudge left over from a turn that landed
+	// while the attendant was talking is answered only if somebody has actually
+	// spoken since; otherwise the last word in the record is the attendant's own
+	// and it would be replying to itself, and then to that.
+	said := a.Said()
+	heard := 0
+	for _, s := range said {
+		if s.Speaker != a.who && s.Text != "" {
+			heard++
+		}
+	}
+	if heard == a.answered {
+		return nil
+	}
+
+	text, err := a.reply(ctx, a.Text())
+	if err != nil {
+		return err
+	}
+	if text == "" {
+		// The model was asked to answer with nothing when the room is not
+		// talking to it. Silence is an answer and costs no floor.
+		a.answered = heard
+		return nil
+	}
+	ogg, err := a.speak(ctx, text)
+	if err != nil {
+		return err
+	}
+
+	// Talking. The floor is empty while the attendant holds it — it is dropped
+	// from the room's ranking and subscribes to nothing of its own — so the
+	// first person to make a sound fills it, and that cancels this.
+	talking, hush := context.WithCancel(ctx)
+	defer hush()
+	go func() {
+		if until(talking, f, busy) == nil {
+			hush()
+		}
+	}()
+	if err := say(talking, ogg); err != nil {
+		return err
+	}
+
+	// Filed whether or not it was heard out: the model must not offer again
+	// what it was in the middle of saying when somebody cut in.
+	a.Spoke(a.who, text, time.Now())
+	a.answered = heard
+	return nil
+}
+
+func free(holder string) bool { return holder == "" }
+func busy(holder string) bool { return holder != "" }
+
+// until waits for the floor to be as want says, or for ctx to end.
+//
+// The floor is read rather than subscribed to: it changes hands on the room's
+// clock and not on ours, and `beat` is the interval the floor already looks at
+// itself on, so this learns nothing later than the floor knows it.
+func until(ctx context.Context, f *Floor, want func(holder string) bool) error {
+	t := time.NewTicker(beat)
+	defer t.Stop()
+	for !want(f.Holder()) {
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		case <-t.C:
+		}
+	}
+	return nil
+}
+
+// reply asks our model what to say about the meeting so far.
+func (a *Answer) reply(ctx context.Context, meeting string) (string, error) {
+	body, _ := json.Marshal(map[string]any{
+		"model":      a.model,
+		"max_tokens": most,
+		"messages": []map[string]string{
+			{"role": "system", "content": a.prompt},
+			{"role": "user", "content": meeting},
+		},
+	})
+	got, err := ask(ctx, a.key, http.MethodPost, a.ai+"/v1/chat/completions", "application/json", body)
+	if err != nil {
+		return "", err
+	}
+	var answered struct {
+		Choices []struct {
+			Message struct {
+				Content string `json:"content"`
+			} `json:"message"`
+		} `json:"choices"`
+	}
+	if err := json.Unmarshal(got, &answered); err != nil {
+		return "", fmt.Errorf("model: %w", err)
+	}
+	if len(answered.Choices) == 0 {
+		return "", nil
+	}
+	return strings.TrimSpace(answered.Choices[0].Message.Content), nil
+}
+
+// speak gives the answer a voice: Ogg/Opus, which is what a room carries and what
+// the attendant publishes without re-encoding.
+//
+// Held in memory rather than streamed from the response, because kokoro
+// synthesizes the whole utterance before it answers — there is nothing arriving
+// to stream — and a track reading a live response body would stall the room on
+// the service instead of on nothing.
+func (a *Answer) speak(ctx context.Context, text string) (io.ReadCloser, error) {
+	body, _ := json.Marshal(map[string]any{
+		"model": "kokoro", "voice": a.voice, "response_format": "opus", "input": text,
+	})
+	got, err := ask(ctx, a.key, http.MethodPost, a.speech+"/v1/audio/speech", "application/json", body)
+	if err != nil {
+		return nil, err
+	}
+	return io.NopCloser(bytes.NewReader(got)), nil
+}

answer_test.go

@@ -0,0 +1,295 @@
+package main
+
+import (
+	"bytes"
+	"context"
+	"io"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+)
+
+// The attendant's own side of a meeting, driven on a clock the test moves: a
+// model that answers, a voice that returns the repo's own Ogg/Opus, and a floor
+// with real speakers on it. What is measured is WHEN it speaks, because an
+// attendant that talks over people is not a participant.
+
+// room is where an answer goes when there is no SFU. It takes what would have
+// been published and, like a real room, holds the attendant for as long as the
+// audio would have taken — so being cut off is something that can happen here.
+type room struct {
+	hold time.Duration // how long the audio plays for
+	done chan struct{} // one signal per answer said
+
+	mu   sync.Mutex
+	said [][]byte
+	cut  []bool
+}
+
+func (r *room) say(ctx context.Context, ogg io.ReadCloser) error {
+	heard, err := io.ReadAll(ogg)
+	if err != nil {
+		return err
+	}
+	cut := false
+	if r.hold > 0 {
+		select {
+		case <-time.After(r.hold):
+		case <-ctx.Done():
+			cut = true
+		}
+	}
+	r.mu.Lock()
+	r.said = append(r.said, heard)
+	r.cut = append(r.cut, cut)
+	r.mu.Unlock()
+
+	select {
+	case r.done <- struct{}{}:
+	default:
+	}
+	return nil
+}
+
+func (r *room) heard() ([][]byte, []bool) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	return append([][]byte(nil), r.said...), append([]bool(nil), r.cut...)
+}
+
+// answering is the attendant's voice against the fake: one URL for the model
+// and for the voice, because the estate is one and it is reached one way.
+func answering(notes *Notes, url string) *Answer {
+	return &Answer{
+		Notes: notes, ai: url, model: "best", speech: url, voice: "af_heart",
+		prompt: "answer", who: "attendant", wake: make(chan struct{}, 1),
+	}
+}
+
+// spoke drives one whole turn: somebody talks, the room names them, and then
+// the room falls quiet — which is what closes their turn and frees the floor.
+func spoke(f *Floor, c *clock, who string) {
+	talk(f, c, time.Second, voice{who, 1})
+	f.Speaking([]string{who})
+	talk(f, c, 500*time.Millisecond, voice{who, 1})
+	c.pass(rest + time.Second)
+	f.Speaking(nil)
+}
+
+// waited reports whether an answer arrived inside d.
+func waited(r *room, d time.Duration) bool {
+	select {
+	case <-r.done:
+		return true
+	case <-time.After(d):
+		return false
+	}
+}
+
+// The floor moves from ana to ben, which ends ana's turn — and the attendant
+// now has something to answer while somebody else is talking. It waits.
+//
+// This is the whole difference between a participant and a nuisance: a turn
+// ending is not an invitation, an empty floor is.
+func TestTheAttendantWaitsForTheFloor(t *testing.T) {
+	s, url := serve(t, 0)
+	s.says("The rebuild finished at four.")
+	notes := Record(url, "whisper", "en")
+	say := answering(notes, url)
+
+	c := started()
+	f := chair(c, say)
+	r := &room{done: make(chan struct{}, 4)}
+
+	ctx, stop := context.WithCancel(context.Background())
+	defer stop()
+	go say.tend(ctx, f, r.say)
+
+	// ana holds the floor, then ben takes it off her: her turn closes and asks
+	// for an answer, and ben is mid-sentence.
+	talk(f, c, time.Second, voice{"ana", 1})
+	f.Speaking([]string{"ana"})
+	both := []voice{{"ana", 1}, {"ben", 2}}
+	talk(f, c, notice, both...)
+	f.Speaking([]string{"ben", "ana"})
+	talk(f, c, grab, both...)
+	f.Speaking([]string{"ben", "ana"})
+	if f.Holder() != "ben" {
+		t.Fatalf("ben holds the floor here, not %q", f.Holder())
+	}
+
+	if waited(r, time.Second) {
+		t.Fatal("the attendant answered ana while ben was talking")
+	}
+
+	// ben stops. Now there is a gap, and a gap is what an answer is for.
+	c.pass(rest + time.Second)
+	f.Speaking(nil)
+	if !waited(r, 10*time.Second) {
+		t.Fatal("the floor was free and the attendant never answered")
+	}
+	said, _ := r.heard()
+	t.Logf("held its answer through ben's turn, then said %d bytes of audio into the gap", len(said[0]))
+}
+
+// Somebody starts talking while the attendant is answering. It stops.
+//
+// The floor is empty while the attendant talks — it is not on the ranking and
+// subscribes to nothing of its own — so the first person to make a sound fills
+// it, and that is the signal. Being interrupted costs a syllable.
+func TestSomebodyTalkingCutsTheAnswerShort(t *testing.T) {
+	s, url := serve(t, 0)
+	s.says("I can pull those numbers up.")
+	notes := Record(url, "whisper", "en")
+	say := answering(notes, url)
+
+	c := started()
+	f := chair(c, say)
+	// A minute of audio: far longer than this test will wait, so an answer that
+	// finishes here finished because it was cut off.
+	r := &room{hold: time.Minute, done: make(chan struct{}, 4)}
+
+	ctx, stop := context.WithCancel(context.Background())
+	defer stop()
+	go say.tend(ctx, f, r.say)
+
+	spoke(f, c, "ana")
+
+	// Wait for the attendant to be talking: the model has answered and the audio
+	// is playing when the service has been asked.
+	for at := time.Now(); len(s.asks()) == 0; {
+		if time.Since(at) > 10*time.Second {
+			t.Fatal("the attendant never started answering")
+		}
+		time.Sleep(20 * time.Millisecond)
+	}
+
+	// ben talks over it.
+	talk(f, c, notice, voice{"ben", 2})
+	f.Speaking([]string{"ben"})
+	if f.Holder() != "ben" {
+		t.Fatalf("ben holds the floor here, not %q", f.Holder())
+	}
+
+	if !waited(r, 5*time.Second) {
+		t.Fatal("ben started talking and the attendant kept going")
+	}
+	_, cut := r.heard()
+	if !cut[0] {
+		t.Fatal("the answer played to the end while ben was talking")
+	}
+	t.Log("stopped mid-answer when ben took the floor")
+}
+
+// A turn that ends while the attendant is talking asks for an answer, and by
+// the time that ask is read the attendant has already spoken. Nobody answers
+// themselves — and an attendant that did would then answer THAT, forever.
+func TestTheAttendantDoesNotAnswerItself(t *testing.T) {
+	s, url := serve(t, 0)
+	s.says("Green across both regions.")
+	notes := Record(url, "whisper", "en")
+	say := answering(notes, url)
+
+	c := started()
+	f := chair(c, say)
+	r := &room{done: make(chan struct{}, 4)}
+
+	ctx, stop := context.WithCancel(context.Background())
+	defer stop()
+	go say.tend(ctx, f, r.say)
+
+	spoke(f, c, "ana")
+	if !waited(r, 10*time.Second) {
+		t.Fatal("ana spoke and the attendant did not answer")
+	}
+
+	// The leftover ask, with nothing said since.
+	say.nudge()
+	if waited(r, time.Second) {
+		t.Fatal("the attendant answered its own answer")
+	}
+	if n := len(s.asks()); n != 1 {
+		t.Fatalf("the model was asked %d times for one thing said", n)
+	}
+	t.Log("one turn, one answer, and the leftover ask went nowhere")
+}
+
+// What the attendant said is in the record under its own name, and it is in
+// what the model reads next. A participant that cannot hear itself repeats
+// itself.
+func TestWhatTheAttendantSaidIsInTheRecord(t *testing.T) {
+	s, url := serve(t, 0)
+	const reply = "Both regions are green."
+	s.says(reply)
+	notes := Record(url, "whisper", "en")
+	say := answering(notes, url)
+
+	c := started()
+	f := chair(c, say)
+	r := &room{done: make(chan struct{}, 4)}
+
+	ctx, stop := context.WithCancel(context.Background())
+	defer stop()
+	go say.tend(ctx, f, r.say)
+
+	spoke(f, c, "ana")
+	if !waited(r, 10*time.Second) {
+		t.Fatal("ana spoke and the attendant did not answer")
+	}
+	spoke(f, c, "ben")
+	if !waited(r, 10*time.Second) {
+		t.Fatal("ben spoke and the attendant did not answer")
+	}
+
+	kept := notes.Text()
+	if !strings.Contains(kept, "attendant: "+reply) {
+		t.Fatalf("the record does not say what the attendant said:\n%s", kept)
+	}
+	asked := s.asks()
+	if len(asked) < 2 {
+		t.Fatalf("the model was asked %d times across two turns", len(asked))
+	}
+	if !strings.Contains(asked[1], reply) {
+		t.Fatalf("the second thing the model read does not contain what it already said:\n%s", asked[1])
+	}
+	t.Logf("the record the model reads on the second turn:\n%s", asked[1])
+}
+
+// What is published is the audio the voice made, and it is audio a room can
+// play: real Ogg/Opus that decodes to speech at the rate the estate speaks in.
+func TestTheAnswerIsAudioTheRoomCanPlay(t *testing.T) {
+	s, url := serve(t, 0)
+	s.says("Say something back.")
+	notes := Record(url, "whisper", "en")
+	say := answering(notes, url)
+
+	c := started()
+	f := chair(c, say)
+	r := &room{done: make(chan struct{}, 4)}
+
+	ctx, stop := context.WithCancel(context.Background())
+	defer stop()
+	go say.tend(ctx, f, r.say)
+
+	spoke(f, c, "ana")
+	if !waited(r, 10*time.Second) {
+		t.Fatal("ana spoke and the attendant did not answer")
+	}
+
+	said, _ := r.heard()
+	pcm := decode(t, bytes.NewReader(said[0]))
+	seconds := float64(len(pcm)) / rate
+	if seconds < 3.5 {
+		t.Fatalf("published %d bytes that decode to %.2fs; the voice answered with 3.9s of speech",
+			len(said[0]), seconds)
+	}
+	var sum float64
+	for _, v := range pcm {
+		sum += float64(v) * float64(v)
+	}
+	if rms := sum / float64(len(pcm)); rms < 200*200 {
+		t.Fatalf("what was published decodes to silence, not speech")
+	}
+	t.Logf("published %d bytes of Ogg/Opus: %.2fs of speech", len(said[0]), seconds)
+}

attend.go

@@ -16,6 +16,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"io"
 	"sync"
 
 	"github.com/livekit/protocol/livekit"
@@ -177,20 +178,33 @@ func (a *Attendant) listen(track *webrtc.TrackRemote, _ *lksdk.RemoteTrackPublic
 // disconnect does not stall every other room's dismissal.
 func (a *Attendant) Leave() { a.left.dismiss() }
 
-// Say publishes audio into the room. The bytes must already be Opus in an Ogg
-// container, which is what hanzoai/speech returns for response_format "opus" —
-// so nothing here encodes, and there is no encoder to keep honest.
-func (a *Attendant) Say(ogg string) error {
-	track, err := lksdk.NewLocalFileTrack(ogg)
+// Say publishes audio into the room and returns when the room has heard all of
+// it — or the moment ctx is cancelled, which is how the attendant is cut off
+// mid-sentence. The track comes down either way: one left published is a
+// microphone nobody takes down, and every answer would add another.
+//
+// The bytes must already be Opus in an Ogg container, which is what
+// hanzoai/speech returns for response_format "opus" — so nothing here encodes,
+// and there is no encoder to keep honest. They are read as they are sent, so
+// there is no file of ours on the way in or a temporary one to clean up.
+func (a *Attendant) Say(ctx context.Context, ogg io.ReadCloser) error {
+	spoken := make(chan struct{})
+	track, err := lksdk.NewLocalReaderTrack(ogg, webrtc.MimeTypeOpus,
+		lksdk.ReaderTrackWithOnWriteComplete(func() { close(spoken) }))
 	if err != nil {
-		return fmt.Errorf("attend: read speech: %w", err)
+		return fmt.Errorf("attend: speech: %w", err)
 	}
-	if _, err := a.room.LocalParticipant.PublishTrack(track, &lksdk.TrackPublicationOptions{
+	pub, err := a.room.LocalParticipant.PublishTrack(track, &lksdk.TrackPublicationOptions{
 		Source: livekit.TrackSource_MICROPHONE,
-	}); err != nil {
+	})
+	if err != nil {
 		return fmt.Errorf("attend: publish: %w", err)
 	}
+	defer a.room.LocalParticipant.UnpublishTrack(pub.SID())
+
+	select {
+	case <-spoken:
+	case <-ctx.Done():
+	}
 	return nil
 }
-
-func main() {}

decode_test.go

@@ -1,6 +1,7 @@
 package main
 
 import (
+	"io"
 	"math"
 	"os"
 	"testing"
@@ -18,8 +19,14 @@ func decodeOgg(t *testing.T, path string) []int16 {
 		t.Skipf("fixture missing: %v", err)
 	}
 	defer f.Close()
+	return decode(t, f)
+}
 
-	ogg, _, err := oggreader.NewWith(f)
+// decode is that conversion over anything Ogg: a fixture on disk, or the bytes
+// an answer was published from.
+func decode(t *testing.T, in io.Reader) []int16 {
+	t.Helper()
+	ogg, _, err := oggreader.NewWith(in)
 	if err != nil {
 		t.Fatalf("ogg: %v", err)
 	}

hanzo.yml

@@ -0,0 +1,28 @@
+# Canonical CI/CD config for hanzoai/attendant — read by the hanzoai/ci reusable
+# (.hanzo/workflows/cicd.yml) and by platform.hanzo.ai.
+#
+# No deploy: here. cd.hanzo.ai rolls the fleet from a pinned tag in
+# hanzo/universe (charts/app/values/hanzo/attendant.yaml), so a build publishes
+# an image and nothing moves until that pin does.
+images:
+  - name: attendant
+    context: .
+    dockerfile: Dockerfile
+    repo: ghcr.io/hanzoai/attendant
+
+# The suite under -race, because everything here is concurrent by nature: audio
+# arrives on one goroutine per speaker, the floor changes hands on another, and
+# the answer is formed on a third.
+#
+# ./mutate.sh is the other half of this gate and it is deliberately NOT run
+# here. It breaks each fix IN PLACE and restores it, and this runner builds the
+# image from the same working tree — a run killed between those two steps (a
+# runner takes SIGTERM on every rolling update, which is why it is given half an
+# hour to finish) would leave the mutation in the tree for the build step to
+# compile and publish. It is run before pushing instead, where the only thing it
+# can damage is a checkout.
+test:
+  - name: go-vet
+    run: go vet ./...
+  - name: go-test
+    run: go test -race -count=1 ./...

meet.go

@@ -0,0 +1,196 @@
+// The way in.
+//
+// The attendant holds no LiveKit signing key. It asks cloud's meet app for one
+// room's token exactly as a browser does — the lobby for where the SFU is, then
+// the token itself — so what it carries is one room's capability, and the key
+// that could open every room in the deployment stays where it already is.
+//
+// It carries ONE identity, from Hanzo IAM, and carries it to all of them: meet,
+// the model, and the voice. Nothing here mints anything.
+//
+// meet admits a PERSON. It wants a subject and a privileged membership in the
+// room's workspace, and refuses a machine credential by name — so the
+// attendant's identity is an IAM account like anyone else's, a member of the
+// workspace it sits in, and the room it joins is named the way meet names
+// rooms: <workspace>_<room>_<id>.
+package main
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"log"
+	"net/http"
+	"os"
+	"os/signal"
+	"strings"
+	"syscall"
+	"time"
+
+	lksdk "github.com/livekit/server-sdk-go/v2"
+)
+
+// What the model is told it is.
+//
+// It answers into a room where people talk to each other as well as to it, so
+// the first thing it needs is permission to say nothing: no words at all is how
+// it stays out of a conversation that is not with it, and the only way it has
+// of not taking a floor it was not offered.
+const brief = "You are the attendant, a participant in a spoken meeting. " +
+	"You are given the transcript so far, one line per turn, with names on them. " +
+	"Answer the last thing said, in two sentences at most, plainly, as speech: " +
+	"no lists, no markdown, no stage directions. " +
+	"If the last thing said was not addressed to you, answer with nothing at all."
+
+// reach is the one client the attendant calls anything with. `patience` bounds
+// every call for the reason it bounds an answer: a call that never returns is a
+// meeting that never hears from the attendant again.
+var reach = &http.Client{Timeout: patience}
+
+// ask sends one request with the attendant's identity on it and answers with
+// the body.
+func ask(ctx context.Context, key, method, url, mime string, body []byte) ([]byte, error) {
+	var in io.Reader
+	if body != nil {
+		in = bytes.NewReader(body)
+	}
+	req, err := http.NewRequestWithContext(ctx, method, url, in)
+	if err != nil {
+		return nil, err
+	}
+	if mime != "" {
+		req.Header.Set("Content-Type", mime)
+	}
+	if key != "" {
+		req.Header.Set("Authorization", "Bearer "+key)
+	}
+	res, err := reach.Do(req)
+	if err != nil {
+		return nil, err
+	}
+	defer res.Body.Close()
+	got, err := io.ReadAll(res.Body)
+	if err != nil {
+		return nil, err
+	}
+	if res.StatusCode >= 300 {
+		return nil, fmt.Errorf("%s %s: %s: %s", method, url, res.Status, bytes.TrimSpace(got))
+	}
+	return got, nil
+}
+
+// seat is what meet's lobby says: who the caller is in a room, and where the
+// rooms are. The identity is meet's to decide and not ours to choose, which is
+// why it is read back rather than configured.
+type seat struct {
+	Identity string `json:"identity"`
+	WS       string `json:"ws"`
+}
+
+func lobby(ctx context.Context, meet, key string) (seat, error) {
+	got, err := ask(ctx, key, http.MethodGet, meet+"/v1/meet/session", "", nil)
+	if err != nil {
+		return seat{}, err
+	}
+	var here seat
+	if err := json.Unmarshal(got, &here); err != nil {
+		return seat{}, err
+	}
+	return here, nil
+}
+
+// ticket is the join token for one room. It answers in text, not JSON, because
+// that is what meet answers with — the token is the whole body.
+func ticket(ctx context.Context, meet, key, room, name string) (string, error) {
+	body, _ := json.Marshal(map[string]string{"roomName": room, "participantName": name})
+	got, err := ask(ctx, key, http.MethodPost, meet+"/v1/meet/getToken", "application/json", body)
+	if err != nil {
+		return "", err
+	}
+	return strings.TrimSpace(string(got)), nil
+}
+
+// env is a setting with a default. Everything past the room and the identity
+// has one.
+func env(name, or string) string {
+	if v := strings.TrimSpace(os.Getenv(name)); v != "" {
+		return v
+	}
+	return or
+}
+
+func main() {
+	room, key := env("ROOM", ""), env("HANZO_TOKEN", "")
+	if room == "" || key == "" {
+		log.Fatal("ROOM and HANZO_TOKEN are required")
+	}
+	meet := env("MEET", "https://api.hanzo.ai")
+
+	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
+	defer stop()
+
+	here, err := lobby(ctx, meet, key)
+	if err != nil {
+		log.Fatalf("meet: %v", err)
+	}
+	if here.WS == "" {
+		log.Fatal("meet names no SFU: LIVEKIT_WS is unset where meet runs")
+	}
+	token, err := ticket(ctx, meet, key, room, env("NAME", "Attendant"))
+	if err != nil {
+		log.Fatalf("meet: %v", err)
+	}
+
+	speech := env("SPEECH", "http://speech.hanzo.svc")
+	notes := Record(speech, "whisper", "en")
+	say := &Answer{
+		Notes:  notes,
+		ai:     env("AI", "https://api.hanzo.ai"),
+		model:  env("MODEL", "best"),
+		speech: speech,
+		voice:  env("VOICE", "af_heart"),
+		prompt: env("PROMPT", brief),
+		who:    here.Identity,
+		key:    key,
+		wake:   make(chan struct{}, 1),
+	}
+
+	floor := Chair(ctx, say)
+	att, err := Attend(ctx, here.WS, token, floor)
+	if err != nil {
+		log.Fatalf("join %s: %v", room, err)
+	}
+	defer att.Leave()
+	log.Printf("attending %s as %s", room, here.Identity)
+
+	go adjourn(ctx, att, stop)
+	say.tend(ctx, floor, att.Say)
+
+	if err := floor.Close(); err != nil {
+		log.Printf("close: %v", err)
+	}
+	fmt.Print(notes.Text())
+}
+
+// adjourn ends the process when the membership does. A join token is good for ten
+// minutes and this one is spent, so the way back into a room we have fallen out
+// of is a new process with a new token — not this one sitting in a room it can
+// no longer hear.
+func adjourn(ctx context.Context, a *Attendant, gone func()) {
+	t := time.NewTicker(beat)
+	defer t.Stop()
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-t.C:
+			if a.room.ConnectionState() == lksdk.ConnectionStateDisconnected {
+				log.Print("the room is gone")
+				gone()
+				return
+			}
+		}
+	}
+}

mutate.sh

@@ -137,4 +137,36 @@ try 'audio given to two turns' TestAudioIsGivenToOneTurnOnly floor.go \
 try 'wait starts at the report' TestTheWaitRunsFromWhenTheyStartedTalking floor.go \
 	'f.rival, f.from = top, f.began(top, now)' 'f.rival, f.from = top, now'
 
+
+# A session lives in one pod's memory and the Service address in front of it
+# round-robins: ignoring where the open says it lives is a 404 for half a turn.
+try 'session addressed at the service' TestASessionIsAddressedWhereItLives scribe.go \
+	'	if got.At == "" {
+		return got.ID, n.url, nil
+	}
+	return got.ID, got.At, nil' '	return got.ID, n.url, nil'
+
+try 'answers over the speaker' TestTheAttendantWaitsForTheFloor answer.go \
+	'	if err := until(ctx, f, free); err != nil {
+		return err
+	}' '	_ = free'
+
+try 'answer cannot be interrupted' TestSomebodyTalkingCutsTheAnswerShort answer.go \
+	'	go func() {
+		if until(talking, f, busy) == nil {
+			hush()
+		}
+	}()' '	_ = busy'
+
+try 'answers itself' TestTheAttendantDoesNotAnswerItself answer.go \
+	'	if heard == a.answered {
+		return nil
+	}' '	_ = heard'
+
+try 'forgets what it said' TestWhatTheAttendantSaidIsInTheRecord answer.go \
+	'	a.Spoke(a.who, text, time.Now())' '	_ = text'
+
+try 'publishes a container the room cannot carry' TestTheAnswerIsAudioTheRoomCanPlay answer.go \
+	'"response_format": "opus"' '"response_format": "mp3"'
+
 echo 'every test earns its keep'

scribe.go

@@ -73,7 +73,7 @@ func Record(url, model, lang string) *Notes {
 
 // Turn opens a transcript for speaker.
 func (n *Notes) Turn(speaker string, at time.Time) (Turn, error) {
-	id, err := n.open()
+	id, host, err := n.open()
 	if err != nil {
 		return nil, err
 	}
@@ -82,12 +82,25 @@ func (n *Notes) Turn(speaker string, at time.Time) (Turn, error) {
 	n.next++
 	n.mu.Unlock()
 
-	t := &turn{to: n, seq: seq, speaker: speaker, at: at, id: id,
+	t := &turn{to: n, seq: seq, speaker: speaker, at: at, id: id, host: host,
 		in: make(chan []int16, queue), gone: make(chan struct{})}
 	go t.push()
 	return t, nil
 }
 
+// Spoke files what the attendant said itself. It is a participant like the rest
+// of them, and a record that leaves one out is not the meeting — this is what
+// the attendant reads back to know what it has already answered.
+//
+// Seconds stays zero: nothing was transcribed, and Seconds is the transcriber's
+// bill. Ordered like any turn, by when it began, which is when it was said.
+func (n *Notes) Spoke(speaker, text string, at time.Time) {
+	n.mu.Lock()
+	defer n.mu.Unlock()
+	n.said[n.next] = Said{Speaker: speaker, At: at, Text: text}
+	n.next++
+}
+
 // Said is every turn closed so far, in the order the turns began. Turns close
 // out of order — a long one settles after the short one that followed it — so
 // they are ordered by when they opened, which is when they were said.
@@ -139,6 +152,7 @@ type turn struct {
 
 	// Owned by push().
 	id      string
+	host    string // where this session lives; see open()
 	pending []byte
 	sent    float64 // seconds pushed into the current session
 	text    string  // settled text of sessions already rolled over
@@ -204,7 +218,8 @@ func (t *turn) send(pcm []byte) {
 }
 
 // roll closes the session at its limit and opens the next, keeping what the
-// closed one settled. The turn continues; only the session behind it changes.
+// closed one settled. The turn continues; only the session behind it changes —
+// and the next one may live somewhere else, so it is addressed where it says.
 func (t *turn) roll() {
 	said, err := t.close(t.id)
 	if err != nil {
@@ -214,12 +229,12 @@ func (t *turn) roll() {
 	t.text = join(t.text, said.Text)
 	t.total += said.Seconds
 
-	id, err := t.to.open()
+	id, host, err := t.to.open()
 	if err != nil {
 		t.err = err
 		return
 	}
-	t.id, t.sent = id, 0
+	t.id, t.host, t.sent = id, host, 0
 }
 
 // settle closes the session and files the turn.
@@ -241,29 +256,42 @@ func (t *turn) settle() {
 	t.to.mu.Unlock()
 }
 
-func (t *turn) url(id string) string { return t.to.url + "/v1/audio/transcript/" + id }
+func (t *turn) url(id string) string { return t.host + "/v1/audio/transcript/" + id }
 
-// state is what a push or a close answers with.
+// state is what an open, a push or a close answers with.
 type state struct {
 	ID      string  `json:"id"`
+	At      string  `json:"at"`
 	Text    string  `json:"text"`
 	Pending string  `json:"pending"`
 	Seconds float64 `json:"seconds"`
 }
 
-// open begins a session. The shape is stated in full because the service checks
-// it and refuses a mismatch rather than resampling: audio carries no header, so
-// a wrong rate is gibberish, not an error.
-func (n *Notes) open() (string, error) {
+// open begins a session, and answers with where that session lives.
+//
+// A growing transcript is a window in ONE process's memory, and a Service
+// address round-robins per connection — so the pushes after the first must go
+// to the pod that took the session, and the open says which that is. Speech
+// runs two replicas, so half of every turn would come back "no transcript"
+// addressed to the Service instead. An empty answer means there is nothing to
+// pin to (a single process), and the address we already have stands.
+//
+// The shape is stated in full because the service checks it and refuses a
+// mismatch rather than resampling: audio carries no header, so a wrong rate is
+// gibberish, not an error.
+func (n *Notes) open() (id, host string, err error) {
 	body, _ := json.Marshal(map[string]any{
 		"model": n.model, "language": n.lang,
 		"format": "pcm16", "rate": rate, "channels": channels,
 	})
 	got, err := n.post(n.url+"/v1/audio/transcript", bytes.NewReader(body), "application/json")
 	if err != nil {
-		return "", err
+		return "", "", err
 	}
-	return got.ID, nil
+	if got.At == "" {
+		return got.ID, n.url, nil
+	}
+	return got.ID, got.At, nil
 }
 
 func (t *turn) close(id string) (state, error) {

scribe_test.go

@@ -6,14 +6,17 @@ import (
 	"io"
 	"net/http"
 	"net/http/httptest"
+	"os"
 	"strings"
 	"sync"
 	"testing"
 	"time"
 )
 
-// service is hanzoai/speech's transcript API, answering the way it answers: a
-// session per open, a limit past which a push is refused, and the text on close.
+// service is the estate the attendant talks to, answering the way it answers:
+// hanzoai/speech's transcript API — a session per open, a limit past which a
+// push is refused, and the text on close — and beside it the two routes an
+// answer is made of, the model and the voice.
 type service struct {
 	mu      sync.Mutex
 	next    int
@@ -22,6 +25,10 @@ type service struct {
 	refused int              // pushes answered 409
 	hold    chan struct{}    // when non-nil, pushes wait on it
 	limit   float64          // seconds a session accepts before 409, 0 for no limit
+	at      string           // where an open says its session lives, empty for here
+	sent    []string         // the host each push was addressed to
+	reply   string           // what the model answers
+	asked   []string         // and what it was asked, last message first
 }
 
 func serve(t *testing.T, limit float64) (*service, string) {
@@ -37,27 +44,74 @@ func (s *service) handle(w http.ResponseWriter, r *http.Request) {
 	id = strings.TrimPrefix(id, "/")
 
 	switch {
-	case r.Method == http.MethodPost && id == "":
-		var ask map[string]any
-		if err := json.NewDecoder(r.Body).Decode(&ask); err != nil {
+	// The model, and the voice. One service stands in for the whole estate
+	// because the attendant reaches all of it the same way.
+	case r.URL.Path == "/v1/chat/completions":
+		var heard struct {
+			Messages []struct{ Role, Content string } `json:"messages"`
+		}
+		if err := json.NewDecoder(r.Body).Decode(&heard); err != nil {
 			http.Error(w, err.Error(), 400)
 			return
 		}
-		if ask["format"] != "pcm16" || ask["rate"] != float64(rate) || ask["channels"] != float64(1) {
-			http.Error(w, fmt.Sprintf("wrong shape: %v", ask), 400)
+		s.mu.Lock()
+		if n := len(heard.Messages); n > 0 {
+			s.asked = append(s.asked, heard.Messages[n-1].Content)
+		}
+		said := s.reply
+		s.mu.Unlock()
+		json.NewEncoder(w).Encode(map[string]any{
+			"choices": []map[string]any{{"message": map[string]string{"content": said}}},
+		})
+
+	// Real kokoro output, so what the attendant publishes is the audio a room
+	// would actually be given rather than a stand-in that decodes to nothing.
+	case r.URL.Path == "/v1/audio/speech":
+		var want struct {
+			Format string `json:"response_format"`
+		}
+		if err := json.NewDecoder(r.Body).Decode(&want); err != nil {
+			http.Error(w, err.Error(), 400)
+			return
+		}
+		// The service refuses a format it cannot make rather than answering in
+		// another one, so asking for the wrong container is an error and not an
+		// MP3 in an Opus track, which a room carries as nothing at all.
+		if want.Format != "opus" {
+			http.Error(w, "unsupported response_format "+want.Format, 400)
+			return
+		}
+		ogg, err := os.ReadFile("spoken.opus")
+		if err != nil {
+			http.Error(w, err.Error(), 500)
+			return
+		}
+		w.Header().Set("Content-Type", "audio/ogg")
+		w.Write(ogg)
+
+	case r.Method == http.MethodPost && id == "":
+		var open map[string]any
+		if err := json.NewDecoder(r.Body).Decode(&open); err != nil {
+			http.Error(w, err.Error(), 400)
+			return
+		}
+		if open["format"] != "pcm16" || open["rate"] != float64(rate) || open["channels"] != float64(1) {
+			http.Error(w, fmt.Sprintf("wrong shape: %v", open), 400)
 			return
 		}
 		s.mu.Lock()
 		s.next++
 		id = fmt.Sprintf("atr_%d", s.next)
 		s.got[id] = nil
+		at := s.at
 		s.mu.Unlock()
 		w.WriteHeader(201)
-		json.NewEncoder(w).Encode(state{ID: id})
+		json.NewEncoder(w).Encode(state{ID: id, At: at})
 
 	case r.Method == http.MethodPost:
 		body, _ := io.ReadAll(r.Body)
 		s.mu.Lock()
+		s.sent = append(s.sent, r.Host)
 		hold, seen := s.hold, s.got[id]
 		if seen == nil {
 			if _, live := s.got[id]; !live {
@@ -88,6 +142,7 @@ func (s *service) handle(w http.ResponseWriter, r *http.Request) {
 
 	case r.Method == http.MethodDelete:
 		s.mu.Lock()
+		s.sent = append(s.sent, r.Host)
 		pushes, live := s.got[id]
 		if !live {
 			s.mu.Unlock()
@@ -117,6 +172,35 @@ func (s *service) pushes(id string) []int {
 	return append([]int(nil), s.kept[id]...)
 }
 
+// says is what the model will answer. Set through the service's own lock, like
+// everything else it holds, so the race detector has the whole picture.
+func (s *service) says(text string) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.reply = text
+}
+
+// asks is what the model was asked, one entry per ask.
+func (s *service) asks() []string {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return append([]string(nil), s.asked...)
+}
+
+// lives is where an open will say its session lives.
+func (s *service) lives(url string) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.at = url
+}
+
+// addressed is the host each push and close was sent to.
+func (s *service) addressed() []string {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return append([]string(nil), s.sent...)
+}
+
 // taken is the audio the service has actually accepted, in seconds.
 func (s *service) taken() float64 {
 	s.mu.Lock()
@@ -218,6 +302,46 @@ func TestALongTurnRollsOverRatherThanBeingRefused(t *testing.T) {
 	}
 }
 
+// A growing transcript is a window in ONE process's memory, and the Service
+// address in front of it round-robins per connection. speech runs two replicas,
+// so a session opened through the Service and pushed through it again lands
+// about half the time on a pod that has never heard of it and answers 404. The
+// open says where the session lives; the rest of the turn goes THERE.
+//
+// Two addresses onto one service, which is what two replicas behind a Service
+// is: the open goes in the front door and names the pod, and every push after
+// it has to be addressed to that pod.
+func TestASessionIsAddressedWhereItLives(t *testing.T) {
+	s, door := serve(t, 0)
+	pod := httptest.NewServer(http.HandlerFunc(s.handle))
+	t.Cleanup(pod.Close)
+	s.lives(pod.URL)
+
+	notes := Record(door, "whisper", "en")
+	turn, err := notes.Turn("ana", time.Now())
+	if err != nil {
+		t.Fatalf("open: %v", err)
+	}
+	turn.Hear(make([]int16, chunk))
+	if err := turn.Close(); err != nil {
+		t.Fatalf("close: %v", err)
+	}
+
+	want := strings.TrimPrefix(pod.URL, "http://")
+	sent := s.addressed()
+	if len(sent) == 0 {
+		t.Fatal("the turn pushed nothing at all")
+	}
+	for _, host := range sent {
+		if host != want {
+			t.Fatalf("a push went to %s; the open said the session lives at %s, "+
+				"and behind two replicas that address is a 404 for half the turn",
+				host, want)
+		}
+	}
+	t.Logf("%d pushes, all addressed to %s", len(sent), want)
+}
+
 // A turn that runs long settles after the short one that followed it. Notes read
 // in the order things were SAID, so they are ordered by when a turn began.
 func TestNotesReadInTheOrderThingsWereSaid(t *testing.T) {