From 8a3ec0bbf92f8da0bdac22a1f2a3f6ccc7202fff Mon Sep 17 00:00:00 2001 From: Jason Lernerman Date: Fri, 25 Sep 2026 11:30:39 -0400 Subject: [PATCH 1/3] refactor(agent): rename debugger textSession to agentSession, extract awaitQuiet Pure refactor ahead of audio mode: the session type is no longer text-specific, and Listen's settle loop becomes a helper a turn can reuse after its speech. --- cmd/lk/debugger_agent.go | 76 +++++++++++++++++++---------------- cmd/lk/debugger_agent_test.go | 14 +++---- cmd/lk/debugger_daemon.go | 4 +- 3 files changed, 51 insertions(+), 43 deletions(-) diff --git a/cmd/lk/debugger_agent.go b/cmd/lk/debugger_agent.go index 21077d09..896162e5 100644 --- a/cmd/lk/debugger_agent.go +++ b/cmd/lk/debugger_agent.go @@ -31,11 +31,11 @@ import ( agent "github.com/livekit/protocol/livekit/agent" ) -// textSession drives an agent connected over the console IPC socket purely in +// agentSession drives an agent connected over the console IPC socket purely in // text mode: it owns the read loop, routes responses to their requests, fans // events out to whoever is listening, and buffers anything nobody was listening // for so it can be reported later instead of silently dropped. -type textSession struct { +type agentSession struct { conn net.Conn reader io.Reader writeMu sync.Mutex @@ -62,13 +62,13 @@ type eventSub struct { ch chan turnEvent } -// newTextSession wraps an accepted agent connection. reader lets the caller +// newAgentSession wraps an accepted agent connection. reader lets the caller // hand back bytes it already peeked from conn (see classifyConn). -func newTextSession(conn net.Conn, reader io.Reader) *textSession { +func newAgentSession(conn net.Conn, reader io.Reader) *agentSession { if reader == nil { reader = conn } - s := &textSession{ + s := &agentSession{ conn: conn, reader: reader, pending: make(map[string]chan *agent.SessionResponse), @@ -83,15 +83,15 @@ func newTextSession(conn net.Conn, reader io.Reader) *textSession { } // Done is closed when the agent connection ends. -func (s *textSession) Done() <-chan struct{} { return s.done } +func (s *agentSession) Done() <-chan struct{} { return s.done } // Turns reports how many RunInput turns have been sent. -func (s *textSession) Turns() int { return int(s.turns.Load()) } +func (s *agentSession) Turns() int { return int(s.turns.Load()) } // TurnInProgress reports whether a RunInput is currently awaiting its reply. -func (s *textSession) TurnInProgress() bool { return len(s.turnSem) > 0 } +func (s *agentSession) TurnInProgress() bool { return len(s.turnSem) > 0 } -func (s *textSession) finish(err error) { +func (s *agentSession) finish(err error) { s.doneOnce.Do(func() { s.mu.Lock() s.readErr = err @@ -104,7 +104,7 @@ func (s *textSession) finish(err error) { }) } -func (s *textSession) readLoop() { +func (s *agentSession) readLoop() { for { msg := &agent.AgentSessionMessage{} if err := ipc.ReadProto(s.reader, msg); err != nil { @@ -141,13 +141,13 @@ func (s *textSession) readLoop() { } } -func (s *textSession) write(msg *agent.AgentSessionMessage) error { +func (s *agentSession) write(msg *agent.AgentSessionMessage) error { s.writeMu.Lock() defer s.writeMu.Unlock() return ipc.WriteProto(s.conn, msg) } -func (s *textSession) notifyActivity() { +func (s *agentSession) notifyActivity() { select { case s.activity <- struct{}{}: default: @@ -157,7 +157,7 @@ func (s *textSession) notifyActivity() { // recentEventsMax bounds the ring of events kept for `events`. const recentEventsMax = 500 -func (s *textSession) handleEvent(ev *agent.AgentSessionEvent) { +func (s *agentSession) handleEvent(ev *agent.AgentSessionEvent) { if ev == nil { return } @@ -195,7 +195,7 @@ func (s *textSession) handleEvent(ev *agent.AgentSessionEvent) { // publishLocked records events in the ring and hands them to observers; when // toTurns is set they also go to turn subscribers, or to the undelivered // buffer if nobody is listening. Caller holds s.mu. -func (s *textSession) publishLocked(events []turnEvent, toTurns bool) { +func (s *agentSession) publishLocked(events []turnEvent, toTurns bool) { s.recent = append(s.recent, events...) if over := len(s.recent) - recentEventsMax; over > 0 { s.recent = append([]turnEvent(nil), s.recent[over:]...) @@ -227,7 +227,7 @@ func (s *textSession) publishLocked(events []turnEvent, toTurns bool) { // observe taps the live event stream without affecting what turns see. It // returns the subscription and the most recent `last` events (all if last <= 0). -func (s *textSession) observe(last int) (*eventSub, []turnEvent) { +func (s *agentSession) observe(last int) (*eventSub, []turnEvent) { obs := &eventSub{ch: make(chan turnEvent, 1024)} s.mu.Lock() recent := s.recent @@ -240,14 +240,14 @@ func (s *textSession) observe(last int) (*eventSub, []turnEvent) { return obs, snapshot } -func (s *textSession) unobserve(obs *eventSub) { +func (s *agentSession) unobserve(obs *eventSub) { s.mu.Lock() delete(s.observers, obs) s.mu.Unlock() } // RecentEvents returns the most recent `last` events (all if last <= 0). -func (s *textSession) RecentEvents(last int) []turnEvent { +func (s *agentSession) RecentEvents(last int) []turnEvent { s.mu.Lock() defer s.mu.Unlock() recent := s.recent @@ -260,7 +260,7 @@ func (s *textSession) RecentEvents(last int) []turnEvent { // subscribe starts receiving live events and, atomically with that, returns // anything that arrived while nobody was listening (marked Earlier) so a turn // can report it before its own output. -func (s *textSession) subscribe() (*eventSub, []turnEvent) { +func (s *agentSession) subscribe() (*eventSub, []turnEvent) { sub := &eventSub{ch: make(chan turnEvent, 256)} s.mu.Lock() earlier := s.undelivered @@ -273,14 +273,14 @@ func (s *textSession) subscribe() (*eventSub, []turnEvent) { return sub, earlier } -func (s *textSession) unsubscribe(sub *eventSub) { +func (s *agentSession) unsubscribe(sub *eventSub) { s.mu.Lock() delete(s.subs, sub) s.mu.Unlock() } // takeUndelivered returns (and clears) events nobody has been shown yet. -func (s *textSession) takeUndelivered() []turnEvent { +func (s *agentSession) takeUndelivered() []turnEvent { s.mu.Lock() earlier := s.undelivered s.undelivered = nil @@ -292,18 +292,18 @@ func (s *textSession) takeUndelivered() []turnEvent { } // AgentState returns the last state the agent reported. -func (s *textSession) AgentState() agent.AgentState { +func (s *agentSession) AgentState() agent.AgentState { s.mu.Lock() defer s.mu.Unlock() return s.agentState } -func (s *textSession) nextRequestID(prefix string) string { +func (s *agentSession) nextRequestID(prefix string) string { return prefix + "-" + strconv.FormatInt(s.reqCounter.Add(1), 10) } // request sends req and waits for its response, ctx, or the connection ending. -func (s *textSession) request(ctx context.Context, req *agent.SessionRequest) (*agent.SessionResponse, error) { +func (s *agentSession) request(ctx context.Context, req *agent.SessionRequest) (*agent.SessionResponse, error) { ch := make(chan *agent.SessionResponse, 1) s.mu.Lock() if s.readErr != nil { @@ -342,7 +342,7 @@ var errAgentGone = errors.New("agent exited") // SetTextMode disables the agent's audio I/O so it runs as a pure text turn // handler, matching what `lk agent console` does when switching to text mode. -func (s *textSession) SetTextMode(ctx context.Context) error { +func (s *agentSession) SetTextMode(ctx context.Context) error { off := false _, err := s.request(ctx, &agent.SessionRequest{ RequestId: s.nextRequestID("io"), @@ -360,7 +360,7 @@ func (s *textSession) SetTextMode(ctx context.Context) error { // on_enter) a chance to finish before the session is reported ready, so the // greeting is shown instead of being lost between turns. Whatever arrives is // left in the undelivered buffer for `start` to print via "pending". -func (s *textSession) WaitForGreeting(ctx context.Context, settle, maxWait time.Duration) { +func (s *agentSession) WaitForGreeting(ctx context.Context, settle, maxWait time.Duration) { s.Listen(ctx, settle, maxWait, func(e turnEvent) { e.Earlier = false s.mu.Lock() @@ -380,7 +380,7 @@ const listenSettle = 300 * time.Millisecond // for the agent to start doing something; once it does, it keeps delivering // until the agent has been quiet for listenSettle, bounded by maxWait overall. // It returns true if any event was delivered. -func (s *textSession) Listen(ctx context.Context, idle, maxWait time.Duration, sink func(turnEvent)) bool { +func (s *agentSession) Listen(ctx context.Context, idle, maxWait time.Duration, sink func(turnEvent)) bool { sub, earlier := s.subscribe() defer s.unsubscribe(sub) @@ -394,11 +394,19 @@ func (s *textSession) Listen(ctx context.Context, idle, maxWait time.Duration, s deadline := time.NewTimer(maxWait) defer deadline.Stop() + return s.awaitQuiet(ctx, sub, delivered, idle, listenSettle, deadline.C, sink) +} + +// awaitQuiet delivers sub's events to sink until the agent goes quiet: settle +// after its last output while it is not busy, or idle if it never produced any +// (delivered reports output seen before the call). It also stops at deadline, +// ctx, or the connection ending, and returns whether any output was delivered. +func (s *agentSession) awaitQuiet(ctx context.Context, sub *eventSub, delivered bool, idle, settle time.Duration, deadline <-chan time.Time, sink func(turnEvent)) bool { quiet := time.NewTimer(idle) defer quiet.Stop() if delivered { // Something was already waiting; only linger for a burst in progress. - quiet.Reset(listenSettle) + quiet.Reset(settle) } busy := false @@ -411,7 +419,7 @@ func (s *textSession) Listen(ctx context.Context, idle, maxWait time.Duration, s } delivered = true if !busy { - quiet.Reset(listenSettle) + quiet.Reset(settle) } case <-s.activity: s.mu.Lock() @@ -424,12 +432,12 @@ func (s *textSession) Listen(ctx context.Context, idle, maxWait time.Duration, s default: if busy { busy = false - quiet.Reset(listenSettle) + quiet.Reset(settle) } } case <-quiet.C: return delivered - case <-deadline.C: + case <-deadline: return delivered case <-ctx.Done(): return delivered @@ -451,7 +459,7 @@ type turnResult struct { // happen, and returns when the agent reports the turn complete. Turns are // serialized; a second caller blocks until the first finishes. Cancel ctx to // stop waiting; any events that arrive afterwards are held for the next turn. -func (s *textSession) Say(ctx context.Context, text string, sink func(turnEvent)) (turnResult, error) { +func (s *agentSession) Say(ctx context.Context, text string, sink func(turnEvent)) (turnResult, error) { select { case s.turnSem <- struct{}{}: defer func() { <-s.turnSem }() @@ -555,7 +563,7 @@ const ( handoffIntroSettle = 300 * time.Millisecond ) -func (s *textSession) awaitHandoffIntro(ctx context.Context, sub *eventSub, deliver func(turnEvent), spoke func() bool) { +func (s *agentSession) awaitHandoffIntro(ctx context.Context, sub *eventSub, deliver func(turnEvent), spoke func() bool) { timer := time.NewTimer(handoffIntroWait) defer timer.Stop() for { @@ -576,7 +584,7 @@ func (s *textSession) awaitHandoffIntro(ctx context.Context, sub *eventSub, deli } // History fetches the agent's authoritative chat history. -func (s *textSession) History(ctx context.Context) ([]turnEvent, error) { +func (s *agentSession) History(ctx context.Context) ([]turnEvent, error) { resp, err := s.request(ctx, &agent.SessionRequest{ RequestId: s.nextRequestID("history"), Request: &agent.SessionRequest_GetChatHistory_{GetChatHistory: &agent.SessionRequest_GetChatHistory{}}, @@ -591,7 +599,7 @@ func (s *textSession) History(ctx context.Context) ([]turnEvent, error) { } // AgentInfo fetches the current agent's id, instructions, and tool names. -func (s *textSession) AgentInfo(ctx context.Context) (*agent.SessionResponse_GetAgentInfoResponse, error) { +func (s *agentSession) AgentInfo(ctx context.Context) (*agent.SessionResponse_GetAgentInfoResponse, error) { resp, err := s.request(ctx, &agent.SessionRequest{ RequestId: s.nextRequestID("info"), Request: &agent.SessionRequest_GetAgentInfo_{GetAgentInfo: &agent.SessionRequest_GetAgentInfo{}}, diff --git a/cmd/lk/debugger_agent_test.go b/cmd/lk/debugger_agent_test.go index f0357e38..6f9d28f9 100644 --- a/cmd/lk/debugger_agent_test.go +++ b/cmd/lk/debugger_agent_test.go @@ -38,7 +38,7 @@ type fakeAgent struct { acks chan struct{} // AudioPlaybackFinished acks from the CLI } -func newFakeAgent(t *testing.T) (*fakeAgent, *textSession) { +func newFakeAgent(t *testing.T) (*fakeAgent, *agentSession) { t.Helper() client, server := net.Pipe() fa := &fakeAgent{t: t, conn: client, reqs: make(chan *agent.SessionRequest, 16), acks: make(chan struct{}, 16)} @@ -57,7 +57,7 @@ func newFakeAgent(t *testing.T) (*fakeAgent, *textSession) { } } }() - sess := newTextSession(server, nil) + sess := newAgentSession(server, nil) t.Cleanup(func() { client.Close(); server.Close() }) return fa, sess } @@ -110,7 +110,7 @@ func userItem(text string) *agent.ChatContext_ChatItem { }}} } -func TestTextSessionSayStreamsEventsAndEarlierItems(t *testing.T) { +func TestAgentSessionSayStreamsEventsAndEarlierItems(t *testing.T) { fa, sess := newFakeAgent(t) // A greeting nobody was listening for must be held, not dropped. @@ -168,7 +168,7 @@ func TestTextSessionSayStreamsEventsAndEarlierItems(t *testing.T) { require.Equal(t, 1, sess.Turns()) } -func TestTextSessionSayReportsAgentErrorAndSilence(t *testing.T) { +func TestAgentSessionSayReportsAgentErrorAndSilence(t *testing.T) { fa, sess := newFakeAgent(t) done := make(chan struct{}) @@ -186,7 +186,7 @@ func TestTextSessionSayReportsAgentErrorAndSilence(t *testing.T) { require.True(t, res.Silent) } -func TestTextSessionSayTimeoutLeavesLateEventsForNextTurn(t *testing.T) { +func TestAgentSessionSayTimeoutLeavesLateEventsForNextTurn(t *testing.T) { fa, sess := newFakeAgent(t) ctx, cancel := context.WithTimeout(context.Background(), 200*time.Millisecond) @@ -207,7 +207,7 @@ func TestTextSessionSayTimeoutLeavesLateEventsForNextTurn(t *testing.T) { require.True(t, earlier[0].Earlier) } -func TestTextSessionRoutesConcurrentResponses(t *testing.T) { +func TestAgentSessionRoutesConcurrentResponses(t *testing.T) { fa, sess := newFakeAgent(t) // A history request issued while a turn is in flight must get its own @@ -244,7 +244,7 @@ func TestTextSessionRoutesConcurrentResponses(t *testing.T) { require.NoError(t, <-sayDone) } -func TestTextSessionAcksPlaybackFlush(t *testing.T) { +func TestAgentSessionAcksPlaybackFlush(t *testing.T) { // Without this ack the agent's turn never completes in text mode. fa, _ := newFakeAgent(t) fa.send(&agent.AgentSessionMessage{Message: &agent.AgentSessionMessage_AudioPlaybackFlush{ diff --git a/cmd/lk/debugger_daemon.go b/cmd/lk/debugger_daemon.go index 8550f19f..07fcec1d 100644 --- a/cmd/lk/debugger_daemon.go +++ b/cmd/lk/debugger_daemon.go @@ -186,7 +186,7 @@ type sessionDaemon struct { lastActive atomic.Int64 // unix nanos of the last control command sessionMu sync.Mutex - session *textSession + session *agentSession agentReady chan struct{} logMu sync.Mutex @@ -262,7 +262,7 @@ func (d *sessionDaemon) handleConn(conn net.Conn) { conn.Close() return } - d.session = newTextSession(conn, reader) + d.session = newAgentSession(conn, reader) d.sessionMu.Unlock() close(d.agentReady) } From a39fa598b40521a0957a05bf585febf35aa2db4d Mon Sep 17 00:00:00 2001 From: Jason Lernerman Date: Fri, 25 Sep 2026 11:45:19 -0400 Subject: [PATCH 2/3] feat(agent): add audio mode to lk agent debugger `lk agent debugger start --audio` speaks each `say` with LiveKit Inference TTS into the agent's microphone input instead of sending text over RunInput, so the agent runs its full audio pipeline (STT, turn detection, TTS, or a realtime model like GPT Live, which only takes audio). Text mode stays the default and is unchanged; the mode is per session, kept across restart, and shown by start and status. In audio mode: - The mic input is a continuous real-time 48 kHz stream: queued speech, else silence, so VAD and endpointing behave as with a live microphone. - Agent audio plays out in real time with no speaker; a playback flush is acknowledged only once that audio would have finished, and a clear cancels it, so agent speaking time and interruptions match a real call. - A turn ends once the agent has transcribed the user, responded, and stayed quiet; a turn it hears but never answers is reported silent. - `say` reports what the agent heard (`heard` in --json) and shows the sent text alongside when the words differ (Sara vs Sarah, 4 vs four). - Project credentials are resolved like other lk commands and passed to the daemon under LK_SESSION_* so the agent keeps its own .env. The echo-agent fixtures gain Inference STT/TTS and the e2e runs both modes. --- .github/workflows/session-e2e.yaml | 5 +- README.md | 6 +- cmd/lk/debugger.go | 114 +++++++-- cmd/lk/debugger_agent.go | 268 +++++++++++++++----- cmd/lk/debugger_agent_test.go | 307 ++++++++++++++++++++++- cmd/lk/debugger_audio.go | 253 +++++++++++++++++++ cmd/lk/debugger_daemon.go | 25 +- cmd/lk/debugger_e2e_test.go | 212 ++++++++-------- cmd/lk/testdata/echo-agent-node/agent.ts | 11 +- cmd/lk/testdata/echo-agent/agent.py | 17 +- cmd/lk/transcript.go | 8 +- go.mod | 2 +- 12 files changed, 1023 insertions(+), 205 deletions(-) create mode 100644 cmd/lk/debugger_audio.go diff --git a/.github/workflows/session-e2e.yaml b/.github/workflows/session-e2e.yaml index f81cbc23..b3f7eb7e 100644 --- a/.github/workflows/session-e2e.yaml +++ b/.github/workflows/session-e2e.yaml @@ -16,8 +16,9 @@ name: Session E2E # Runs on manual dispatch and on pushes to any repo branch (forks can't trigger # `push`, so secrets are only exposed to trusted collaborators). It needs live # LiveKit credentials -- set these repo secrets first: LIVEKIT_API_KEY, -# LIVEKIT_API_SECRET, LIVEKIT_URL. The echo agents drive their LLM through -# LiveKit Inference, so no other provider keys are needed. +# LIVEKIT_API_SECRET, LIVEKIT_URL. The echo agents drive their STT, LLM, and +# TTS through LiveKit Inference (as does the debugger's own TTS in audio mode), +# so no other provider keys are needed. # # Each agent's deps are resolved fresh in CI (they're gitignored): the Python # fixture via `uv sync` from its pyproject.toml; the Node fixture via diff --git a/README.md b/README.md index 4bce89ad..cf0a2fa8 100644 --- a/README.md +++ b/README.md @@ -481,14 +481,16 @@ Console needs a terminal. If a program rather than a person is going to hold the ## Agent debugger -`lk agent debugger` is built for coding agents such as Claude Code, Codex, and Cursor. When one of them is working on your LiveKit agent, it can use this command to hold a text conversation with the agent it just edited, read the replies and tool calls, and decide what to try next, all from the shell, with no human at a microphone. It is an ad-hoc, local stand-in for [Agent Simulations](https://docs.livekit.io/agents/start/testing/simulations/) in which the coding agent plays the user and pays with its own tokens: the agent runs in console mode with STT/TTS disabled, so a turn costs only your LLM and tool calls, and nothing touches a LiveKit room. +`lk agent debugger` is built for coding agents such as Claude Code, Codex, and Cursor. When one of them is working on your LiveKit agent, it can use this command to hold a conversation with the agent it just edited, read the replies and tool calls, and decide what to try next, all from the shell, with no human at a microphone. It is an ad-hoc, local stand-in for [Agent Simulations](https://docs.livekit.io/agents/start/testing/simulations/) in which the coding agent plays the user and pays with its own tokens: by default the agent runs in console mode with STT/TTS disabled, so a turn costs only your LLM and tool calls, and nothing touches a LiveKit room. + +Start the session with `--audio` to test the audio path instead: each `say` is spoken with LiveKit Inference TTS into the agent's microphone input, and the agent runs its full audio pipeline (STT, turn detection, TTS, or a realtime model that only takes audio), so misheard names, spelled-out digits, and endpointing behave as they do on a call. Each turn then prints what the agent transcribed, with the text you sent alongside when the words differ. The TTS uses your project's credentials, resolved like other `lk` commands. People can use it too, but the output and flags are shaped for a program driving it one command at a time: every turn prints tool calls with their arguments and results, handoffs, errors, and the reply; exit codes are non-zero when a turn fails; `--json` is available everywhere. If you want to talk to your agent yourself, `lk agent console` is the better fit. A typical session, run from the agent project directory (`lk agent dbg` is short for `lk agent debugger`): ```shell -lk agent debugger start # starts the agent, prints its opening message if it has one +lk agent debugger start # starts the agent, prints its opening message if it has one (--audio to speak turns) lk agent debugger say "Hi, what can you do?" # prints tool calls (with arguments and results), handoffs, and the reply lk agent debugger say "Book a table for two tonight" lk agent debugger chat-history # the whole conversation so far diff --git a/cmd/lk/debugger.go b/cmd/lk/debugger.go index 50032bfb..94578f18 100644 --- a/cmd/lk/debugger.go +++ b/cmd/lk/debugger.go @@ -26,6 +26,7 @@ import ( "strconv" "strings" "time" + "unicode" "github.com/mattn/go-isatty" "github.com/urfave/cli/v3" @@ -48,6 +49,14 @@ const ( envSessionPType = "LK_SESSION_PTYPE" // agentfs.ProjectType string envSessionReadyFile = "LK_SESSION_READY_FILE" // path the daemon writes its status to envSessionIdle = "LK_SESSION_IDLE_TIMEOUT" // stop after this long without commands (0 = never) + envSessionAudio = "LK_SESSION_AUDIO" // set: speak turns instead of sending text + + // The project credentials an audio session speaks the user's turns with. + // They are namespaced so the agent, which inherits the daemon's + // environment, still resolves its own LIVEKIT_* credentials. + envSessionURL = "LK_SESSION_LIVEKIT_URL" + envSessionAPIKey = "LK_SESSION_LIVEKIT_API_KEY" + envSessionAPISecret = "LK_SESSION_LIVEKIT_API_SECRET" // defaultIdleTimeout is how long the daemon waits for a command before // stopping itself, so a caller that never runs `stop` doesn't leave an @@ -71,6 +80,11 @@ var sessionIdleFlag = &cli.DurationFlag{ Usage: "Stop the session after `DURATION` without any command, such as 30m, 2h, or 90s (0 keeps it running until stop)", } +var sessionAudioFlag = &cli.BoolFlag{ + Name: "audio", + Usage: "Speak each turn into the agent's microphone input instead of sending text, running its full audio pipeline", +} + var sessionMetricsFlag = &cli.BoolFlag{ Name: "metrics", Usage: "Show latency metrics: time to first token per reply and how long each turn took", @@ -98,13 +112,20 @@ func init() { var agentDaemonCommand = &cli.Command{ Name: "debugger", Aliases: []string{"dbg"}, - Usage: "Drive a text conversation with a local agent from a script or coding agent", - Description: `Runs your agent locally in text mode as a background process, then lets -you drive a multi-turn conversation one command at a time. Each "say" sends a -user turn and prints everything the agent did in response: tool calls with -their arguments and results, handoffs, errors, and the reply. Nothing is sent -to a LiveKit room. The agent runs in console mode with STT/TTS disabled, so a -turn costs only what your LLM (and tools) cost. + Usage: "Drive a conversation with a local agent from a script or coding agent", + Description: `Runs your agent locally as a background process, then lets you drive a +multi-turn conversation one command at a time. Each "say" sends a user turn and +prints everything the agent did in response: tool calls with their arguments +and results, handoffs, errors, and the reply. Nothing is sent to a LiveKit room. + +By default the conversation is text: the agent runs in console mode with +STT/TTS disabled, so a turn costs only what your LLM (and tools) cost. Start +with --audio to speak each turn instead: the user's line is spoken with LiveKit +Inference TTS into the agent's microphone input, and the agent runs its full +audio pipeline (STT, turn detection, TTS, or a realtime model that only takes +audio), so misheard names, spelled-out digits, and endpointing show up here. +Audio mode uses your project's credentials for the TTS, resolved like other lk +commands (--project, LIVEKIT_* variables, livekit.toml, or the default project). This is built for coding agents (Claude Code, Codex, Cursor, ...) and shell scripts: the caller stands in for the user, decides the next line based on the @@ -114,7 +135,7 @@ voice or typing, use "lk agent console" instead. Typical flow, run from the agent project directory: - lk agent debugger start # starts the agent, prints its greeting (if any) + lk agent debugger start # starts the agent, prints its greeting (if any); --audio to speak turns lk agent debugger say "Hi, what can you do?" lk agent debugger say "Book me a table for two tonight" lk agent debugger logs --last 40 # agent process logs (tracebacks, warnings) @@ -142,7 +163,7 @@ running. Use --port to run several agents side by side (one session per port). Usage: "Start the agent in the background and wait until it is ready", ArgsUsage: "[entrypoint] [-- node/python-args...]", Description: `Launches the agent as a detached process and returns once it is connected -and ready for text turns. If the agent speaks first (a greeting from on_enter), +and ready for turns, spoken ones with --audio. If the agent speaks first (a greeting from on_enter), that is printed here. The summary line names the active agent and its tools. Without an entrypoint, the project in the current directory (or nearest parent) @@ -156,7 +177,7 @@ The agent reads its own .env for credentials, exactly like "lk agent console". Only one session runs per port; use --port for more, or "restart" to replace the current one. The session stops itself after --idle-timeout (default 30m) without any command, so a forgotten session does not linger.`, - Flags: []cli.Flag{sessionPortFlag, sessionIdleFlag, jsonFlag}, + Flags: []cli.Flag{sessionPortFlag, sessionIdleFlag, sessionAudioFlag, jsonFlag}, Action: runSessionStart, }, { @@ -168,6 +189,11 @@ turn completes: tool calls with their arguments and results, handoffs to other agents, errors, and the reply. Anything the agent said since the previous turn (for example a timer firing) is printed first, marked "(before this turn)". +In an audio session (start --audio) the text is spoken, and the turn lasts until +the agent has heard it, responded, and gone quiet. The user line is what the +agent transcribed, with the text you sent shown alongside when the words +differ; a turn the agent hears but never answers is reported as silent. + Exit code is non-zero if the agent reported an error or the turn timed out; the agent keeps running either way. With --logs, the agent's log lines emitted during the turn are shown beneath the step they belong to, which puts a tool's @@ -301,14 +327,42 @@ func runSessionStart(ctx context.Context, cmd *cli.Command) error { return err } port := int(cmd.Int("port")) + audioEnv, err := sessionAudioEnv(cmd, cmd.Bool("audio")) + if err != nil { + return err + } out.Statusf("Detected %s agent (%s in %s)", projectType.Lang(), util.Accented(entrypoint), util.Accented(projectDir)) - return startSessionDaemon(port, projectDir, projectType, entrypoint, cmd.Duration("idle-timeout"), cmd.Bool("json")) + return startSessionDaemon(port, projectDir, projectType, entrypoint, cmd.Duration("idle-timeout"), audioEnv, cmd.Bool("json")) +} + +func sessionMode(audio bool) string { + if audio { + return "audio" + } + return "text" +} + +// sessionAudioEnv returns the daemon environment for an audio session: the +// mode, plus the project credentials the user's turns are spoken with. +func sessionAudioEnv(cmd *cli.Command, audio bool) ([]string, error) { + if !audio { + return nil, nil + } + creds, err := resolveCredentials(cmd) + if err != nil { + return nil, fmt.Errorf("--audio speaks your turns with LiveKit Inference, which needs project credentials: %w", err) + } + env := []string{envSessionAudio + "=1"} + for _, kv := range creds { + env = append(env, "LK_SESSION_"+kv) + } + return env, nil } // startSessionDaemon launches the detached daemon, waits for it to report // ready, then prints the agent's opening message (if it produced one) and a // short status line. -func startSessionDaemon(port int, projectDir string, projectType agentfs.ProjectType, entrypoint string, idleTimeout time.Duration, asJSON bool) error { +func startSessionDaemon(port int, projectDir string, projectType agentfs.ProjectType, entrypoint string, idleTimeout time.Duration, audioEnv []string, asJSON bool) error { exe, err := os.Executable() if err != nil { return fmt.Errorf("could not resolve own binary: %w", err) @@ -340,6 +394,7 @@ func startSessionDaemon(port int, projectDir string, projectType agentfs.Project envSessionPType+"="+string(projectType), envSessionReadyFile+"="+readyPath, ) + daemon.Env = append(daemon.Env, audioEnv...) daemon.Stdout = logFile daemon.Stderr = logFile setDetachedProcAttr(daemon) @@ -385,7 +440,7 @@ func startSessionDaemon(port int, projectDir string, projectType agentfs.Project summary := "Session started." if st != nil && st.Status != nil { s := st.Status - summary = fmt.Sprintf("Session started. Listening on %s, agent pid %d.", sessionAddr(s.Port), s.Pid) + summary = fmt.Sprintf("Session started in %s mode. Listening on %s, agent pid %d.", sessionMode(s.Audio), sessionAddr(s.Port), s.Pid) if s.AgentID != "" { summary += fmt.Sprintf(" Agent %q", s.AgentID) if len(s.Tools) > 0 { @@ -449,6 +504,7 @@ func readReadyStatus(path string) (string, bool) { // sayJSON is the --json document `say` prints for one turn. type sayJSON struct { Text string `json:"text"` + Heard string `json:"heard"` // what the agent took the user to say: the text, or its transcript of the speech Events []turnEvent `json:"events"` Reply string `json:"reply"` Silent bool `json:"silent,omitempty"` @@ -490,7 +546,6 @@ func runSessionSay(ctx context.Context, cmd *cli.Command) error { asJSON := cmd.Bool("json") opts := renderFlags(cmd) doc := sayJSON{Text: text, Events: []turnEvent{}} - sawAgentOutput := false // Agent log lines are emitted while the agent works on what comes next // (a tool executing, a reply generating), so they are held and printed @@ -508,9 +563,6 @@ func runSessionSay(ctx context.Context, cmd *cli.Command) error { return } e := *r.Event - if e.Type != "log" && (e.Type != "message" || e.Role != "user") && !e.Earlier { - sawAgentOutput = true - } if asJSON { doc.Events = append(doc.Events, e) return @@ -522,6 +574,9 @@ func runSessionSay(ctx context.Context, cmd *cli.Command) error { if line := renderTurnEvent(e, opts); line != "" { out.Result(line) } + if e.Type == "message" && e.Role == "user" && !e.Earlier && !sameWords(e.Text, text) { + out.Result(" " + transcriptDim.Render(fmt.Sprintf("(sent: %q)", text))) + } flushLogs() }) if err != nil { @@ -532,6 +587,7 @@ func runSessionSay(ctx context.Context, cmd *cli.Command) error { } doc.Reply = final.Reply + doc.Heard = final.Heard doc.Silent = final.Silent doc.Error = final.Error doc.DurationMs = final.DurationMs @@ -545,7 +601,10 @@ func runSessionSay(ctx context.Context, cmd *cli.Command) error { return nil } - if final.Error == "" && !sawAgentOutput { + if final.Error == "" && final.Heard == "" { + out.Result(" " + transcriptDim.Render("(the agent transcribed no speech this turn)")) + } + if final.Error == "" && final.Silent { out.Result(renderSilentTurn()) } if opts.Metrics { @@ -558,6 +617,17 @@ func runSessionSay(ctx context.Context, cmd *cli.Command) error { return nil } +// sameWords reports whether a and b say the same words, ignoring case and +// punctuation, so only a real transcription difference is flagged. +func sameWords(a, b string) bool { + words := func(s string) string { + return strings.Join(strings.FieldsFunc(strings.ToLower(s), func(r rune) bool { + return !unicode.IsLetter(r) && !unicode.IsDigit(r) + }), " ") + } + return words(a) == words(b) +} + func runSessionEvents(ctx context.Context, cmd *cli.Command) error { conn, err := dialControl(int(cmd.Int("port"))) if err != nil { @@ -644,6 +714,7 @@ func runSessionStatus(ctx context.Context, cmd *cli.Command) error { label := func(s string) string { return util.Dimmed(fmt.Sprintf(" %-14s", s)) } out.Resultf("Session running on %s (pid %d, up %s)\n", sessionAddr(st.Port), st.Pid, (time.Duration(st.UptimeSeconds) * time.Second).String()) out.Resultf("%s%s (%s, %s)\n", label("Project:"), st.ProjectDir, agentfs.ProjectType(st.ProjectType).Lang(), st.Entrypoint) + out.Resultf("%s%s\n", label("Mode:"), sessionMode(st.Audio)) agentLine := st.AgentID if agentLine == "" { agentLine = util.Dimmed("(unknown)") @@ -823,9 +894,13 @@ func runSessionRestart(ctx context.Context, cmd *cli.Command) error { c.Close() time.Sleep(100 * time.Millisecond) } + audioEnv, err := sessionAudioEnv(cmd, st.Audio) + if err != nil { + return err + } out.Statusf("Restarting %s agent (%s in %s)", agentfs.ProjectType(st.ProjectType).Lang(), st.Entrypoint, st.ProjectDir) return startSessionDaemon(port, st.ProjectDir, agentfs.ProjectType(st.ProjectType), st.Entrypoint, - time.Duration(st.IdleTimeoutSeconds)*time.Second, cmd.Bool("json")) + time.Duration(st.IdleTimeoutSeconds)*time.Second, audioEnv, cmd.Bool("json")) } // dialControl connects to the session daemon and sends the control preamble. @@ -924,6 +999,7 @@ type controlReply struct { Done bool `json:"done,omitempty"` Error string `json:"error,omitempty"` Reply string `json:"reply,omitempty"` // say: concatenated assistant text + Heard string `json:"heard,omitempty"` // say: what the agent took the user to say Silent bool `json:"silent,omitempty"` // say: the agent produced no output DurationMs int64 `json:"duration_ms,omitempty"` } diff --git a/cmd/lk/debugger_agent.go b/cmd/lk/debugger_agent.go index 896162e5..7df2bed8 100644 --- a/cmd/lk/debugger_agent.go +++ b/cmd/lk/debugger_agent.go @@ -26,15 +26,18 @@ import ( "sync/atomic" "time" + "github.com/livekit/livekit-cli/v2/pkg/console" "github.com/livekit/livekit-cli/v2/pkg/ipc" agent "github.com/livekit/protocol/livekit/agent" ) -// agentSession drives an agent connected over the console IPC socket purely in -// text mode: it owns the read loop, routes responses to their requests, fans -// events out to whoever is listening, and buffers anything nobody was listening -// for so it can be reported later instead of silently dropped. +// agentSession drives an agent connected over the console IPC socket: it owns +// the read loop, routes responses to their requests, fans events out to +// whoever is listening, and buffers anything nobody was listening for so it can +// be reported later instead of silently dropped. In text mode the user's turns +// are sent as text with the agent's audio I/O off; in audio mode they are +// spoken into its microphone input and its replies play out in real time. type agentSession struct { conn net.Conn reader io.Reader @@ -49,7 +52,12 @@ type agentSession struct { agentState agent.AgentState activity chan struct{} - turnSem chan struct{} // one RunInput in flight at a time + speak speakFunc // nil in text mode + replyWait time.Duration + mic micStream + speaker playout + + turnSem chan struct{} // one turn in flight at a time turns atomic.Int64 reqCounter atomic.Int64 @@ -63,14 +71,17 @@ type eventSub struct { } // newAgentSession wraps an accepted agent connection. reader lets the caller -// hand back bytes it already peeked from conn (see classifyConn). -func newAgentSession(conn net.Conn, reader io.Reader) *agentSession { +// hand back bytes it already peeked from conn (see classifyConn). A nil speak +// runs the session in text mode. +func newAgentSession(conn net.Conn, reader io.Reader, speak speakFunc) *agentSession { if reader == nil { reader = conn } s := &agentSession{ conn: conn, reader: reader, + speak: speak, + replyWait: defaultReplyWait, pending: make(map[string]chan *agent.SessionResponse), subs: make(map[*eventSub]struct{}), observers: make(map[*eventSub]struct{}), @@ -79,16 +90,31 @@ func newAgentSession(conn net.Conn, reader io.Reader) *agentSession { done: make(chan struct{}), } go s.readLoop() + if speak == nil { + return s + } + go s.mic.run(s.done, func(samples []int16) error { + return s.write(&agent.AgentSessionMessage{ + Message: &agent.AgentSessionMessage_AudioInput{ + AudioInput: &agent.AgentSessionMessage_ConsoleIO_AudioFrame{ + Data: console.SamplesToBytes(samples), + SampleRate: console.SampleRate, + NumChannels: console.Channels, + SamplesPerChannel: uint32(len(samples)), + }, + }, + }) + }) return s } // Done is closed when the agent connection ends. func (s *agentSession) Done() <-chan struct{} { return s.done } -// Turns reports how many RunInput turns have been sent. +// Turns reports how many user turns have been sent. func (s *agentSession) Turns() int { return int(s.turns.Load()) } -// TurnInProgress reports whether a RunInput is currently awaiting its reply. +// TurnInProgress reports whether a turn is currently awaiting its reply. func (s *agentSession) TurnInProgress() bool { return len(s.turnSem) > 0 } func (s *agentSession) finish(err error) { @@ -127,15 +153,19 @@ func (s *agentSession) readLoop() { close(ch) } } - case *agent.AgentSessionMessage_AudioOutput, *agent.AgentSessionMessage_AudioPlaybackClear: - // No audio sink in text mode: drop. + case *agent.AgentSessionMessage_AudioOutput: + s.speaker.write(m.AudioOutput.GetSamplesPerChannel(), m.AudioOutput.GetSampleRate()) + case *agent.AgentSessionMessage_AudioPlaybackClear: + s.speaker.clear() case *agent.AgentSessionMessage_AudioPlaybackFlush: - // Nothing to drain, so ack immediately or the agent's turn (and the - // RunInputResponse we await) never completes. - _ = s.write(&agent.AgentSessionMessage{ - Message: &agent.AgentSessionMessage_AudioPlaybackFinished{ - AudioPlaybackFinished: &agent.AgentSessionMessage_ConsoleIO_AudioPlaybackFinished{}, - }, + // The agent's speech only completes once this is acknowledged. In + // text mode nothing was buffered, so it is acknowledged at once. + s.speaker.flush(func() { + _ = s.write(&agent.AgentSessionMessage{ + Message: &agent.AgentSessionMessage_AudioPlaybackFinished{ + AudioPlaybackFinished: &agent.AgentSessionMessage_ConsoleIO_AudioPlaybackFinished{}, + }, + }) }) } } @@ -387,62 +417,65 @@ func (s *agentSession) Listen(ctx context.Context, idle, maxWait time.Duration, // Markers (the initial agent handoff, config updates) are delivered but do // not count as the agent speaking, so they never shorten the wait. delivered := false - for _, e := range earlier { + deliver := func(e turnEvent) { sink(e) - delivered = delivered || e.isOutput() + delivered = delivered || e.isAgentOutput() + } + for _, e := range earlier { + deliver(e) } deadline := time.NewTimer(maxWait) defer deadline.Stop() - return s.awaitQuiet(ctx, sub, delivered, idle, listenSettle, deadline.C, sink) + s.awaitQuiet(ctx, sub, idle, listenSettle, deadline.C, deliver, func() bool { return delivered }) + return delivered } -// awaitQuiet delivers sub's events to sink until the agent goes quiet: settle -// after its last output while it is not busy, or idle if it never produced any -// (delivered reports output seen before the call). It also stops at deadline, -// ctx, or the connection ending, and returns whether any output was delivered. -func (s *agentSession) awaitQuiet(ctx context.Context, sub *eventSub, delivered bool, idle, settle time.Duration, deadline <-chan time.Time, sink func(turnEvent)) bool { - quiet := time.NewTimer(idle) +func isBusy(state agent.AgentState) bool { + return state == agent.AgentState_AS_THINKING || state == agent.AgentState_AS_SPEAKING +} + +// awaitQuiet hands sub's events to deliver until the agent goes quiet: settle +// after it last produced output or stopped being busy once responded reports +// true, or idle while it has not. It also stops at deadline, ctx, or the +// connection ending. +func (s *agentSession) awaitQuiet(ctx context.Context, sub *eventSub, idle, settle time.Duration, deadline <-chan time.Time, deliver func(turnEvent), responded func() bool) { + wait := func() time.Duration { + if responded() { + return settle + } + return idle + } + quiet := time.NewTimer(wait()) defer quiet.Stop() - if delivered { - // Something was already waiting; only linger for a burst in progress. - quiet.Reset(settle) + busy := isBusy(s.AgentState()) + if busy { + quiet.Stop() } - - busy := false for { select { case e := <-sub.ch: - sink(e) - if !e.isOutput() { - continue - } - delivered = true - if !busy { + deliver(e) + if !busy && responded() { quiet.Reset(settle) } case <-s.activity: - s.mu.Lock() - state := s.agentState - s.mu.Unlock() - switch state { - case agent.AgentState_AS_THINKING, agent.AgentState_AS_SPEAKING: + switch { + case isBusy(s.AgentState()): busy = true quiet.Stop() - default: - if busy { - busy = false - quiet.Reset(settle) - } + case busy: + busy = false + quiet.Reset(wait()) } case <-quiet.C: - return delivered + return case <-deadline: - return delivered + return case <-ctx.Done(): - return delivered + return case <-s.done: - return delivered + return } } } @@ -450,15 +483,29 @@ func (s *agentSession) awaitQuiet(ctx context.Context, sub *eventSub, delivered // turnResult is what Say reports once the agent's turn completes. type turnResult struct { Reply string // concatenated assistant text produced during the turn - Silent bool // the turn completed without any agent output + Heard string // what the agent transcribed from the user's speech + Silent bool // the agent produced no output after hearing the user Duration time.Duration } -// Say runs one user turn: it reports events nobody has seen yet, echoes the -// user text, sends RunInput, streams the agent's events to sink as they -// happen, and returns when the agent reports the turn complete. Turns are -// serialized; a second caller blocks until the first finishes. Cancel ctx to -// stop waiting; any events that arrive afterwards are held for the next turn. +const ( + // defaultReplyWait is how long a turn waits, once the user's speech has + // played and the agent is not busy, for it to transcribe and respond before + // calling the turn silent. It covers endpointing plus the first LLM round trip. + defaultReplyWait = 10 * time.Second + // turnSettle is how long the agent must stay quiet after responding before + // the turn is over. A delegating model (GPT Live) acknowledges, then speaks + // the delegated answer after a pause with no session event marking it, so + // this has to outlast that pause. + turnSettle = 3 * time.Second +) + +// Say runs one user turn: it reports events nobody has seen yet, hands the +// text to the agent (typed in text mode, spoken into its microphone input in +// audio mode), streams the agent's events to sink as they happen, and returns +// when the turn is over. Turns are serialized; a second caller blocks until the +// first finishes. Cancel ctx to stop waiting; any events that arrive afterwards +// are held for the next turn. func (s *agentSession) Say(ctx context.Context, text string, sink func(turnEvent)) (turnResult, error) { select { case s.turnSem <- struct{}{}: @@ -475,6 +522,16 @@ func (s *agentSession) Say(ctx context.Context, text string, sink func(turnEvent for _, e := range earlier { sink(e) } + s.turns.Add(1) + if s.speak == nil { + return s.sayText(ctx, text, start, sub, sink) + } + return s.sayAudio(ctx, text, start, sub, sink) +} + +// sayText echoes the user text, sends RunInput, and returns when the agent +// reports the turn complete. +func (s *agentSession) sayText(ctx context.Context, text string, start time.Time, sub *eventSub, sink func(turnEvent)) (turnResult, error) { sink(turnEvent{Type: "message", Role: "user", Text: text}) var ( @@ -502,7 +559,6 @@ func (s *agentSession) Say(ctx context.Context, text string, sink func(turnEvent sink(e) } - s.turns.Add(1) respCh := make(chan struct { resp *agent.SessionResponse err error @@ -541,7 +597,7 @@ func (s *agentSession) Say(ctx context.Context, text string, sink func(turnEvent if r.err == nil && sawHandoff && !spokeAfter { s.awaitHandoffIntro(ctx, sub, deliver, func() bool { return spokeAfter }) } - res := turnResult{Reply: reply.String(), Silent: !sawAgent, Duration: time.Since(start)} + res := turnResult{Reply: reply.String(), Heard: text, Silent: !sawAgent, Duration: time.Since(start)} if r.err != nil { if errors.Is(r.err, context.DeadlineExceeded) { return res, fmt.Errorf("timed out waiting for the agent's reply; the turn may still be running (check `lk agent debugger logs`)") @@ -556,6 +612,100 @@ func (s *agentSession) Say(ctx context.Context, text string, sink func(turnEvent } } +// sayAudio speaks the text into the agent's microphone input and returns once +// the agent has heard it, responded, and gone quiet, or never responded. +func (s *agentSession) sayAudio(ctx context.Context, text string, start time.Time, sub *eventSub, sink func(turnEvent)) (turnResult, error) { + var ( + reply, heard strings.Builder + agentErr string + responded bool // the agent produced output after hearing the user + sawHandoff bool + spokeAfter bool // an assistant message arrived after the last handoff + ) + appendLine := func(b *strings.Builder, line string) { + if b.Len() > 0 { + b.WriteString("\n") + } + b.WriteString(line) + } + deliver := func(e turnEvent) { + switch { + case e.Type == "message" && e.Role == "user": + appendLine(&heard, e.Text) + case e.Type == "handoff": + sawHandoff = true + spokeAfter = false + case e.Type == "message" && e.Role == "assistant": + spokeAfter = true + appendLine(&reply, e.Text) + case e.Type == "error" && agentErr == "": + agentErr = e.Text + } + if heard.Len() > 0 && e.isAgentOutput() { + responded = true + } + sink(e) + } + result := func() turnResult { + return turnResult{Reply: reply.String(), Heard: heard.String(), Silent: !responded, Duration: time.Since(start)} + } + + spoken := make(chan error, 1) + go func() { + if err := s.speak(ctx, text, s.mic.push); err != nil { + spoken <- err + return + } + select { + case <-s.mic.drained(): + case <-ctx.Done(): + } + spoken <- nil + }() +speaking: + for { + select { + case e := <-sub.ch: + deliver(e) + case err := <-spoken: + if err != nil { + return result(), err + } + break speaking + case <-s.done: + return result(), errAgentGone + } + } + + // Output while the user was still being heard (the tail of an earlier + // reply) does not answer this turn, so the turn waits for the agent to + // transcribe the speech and respond to it. + s.awaitQuiet(ctx, sub, s.replyWait, turnSettle, nil, deliver, func() bool { return responded }) + // A handoff target usually introduces itself from on_enter, which can + // start after the previous agent went quiet. Hold the turn open briefly so + // that introduction reads as part of the handoff. + if ctx.Err() == nil && sawHandoff && !spokeAfter { + s.awaitHandoffIntro(ctx, sub, deliver, func() bool { return spokeAfter }) + } + + res := result() + select { + case <-s.done: + return res, errAgentGone + default: + } + if errors.Is(ctx.Err(), context.DeadlineExceeded) { + return res, fmt.Errorf("timed out waiting for the agent's reply; the turn may still be running (check `lk agent debugger logs`)") + } + if err := ctx.Err(); err != nil { + return res, err + } + if agentErr != "" { + return res, fmt.Errorf("agent error: %s", agentErr) + } + return res, nil +} + // handoffIntroWait bounds how long Say waits for a new agent's opening line // after a handoff; handoffIntroSettle is the quiet period after it arrives. const ( diff --git a/cmd/lk/debugger_agent_test.go b/cmd/lk/debugger_agent_test.go index 6f9d28f9..bc15818f 100644 --- a/cmd/lk/debugger_agent_test.go +++ b/cmd/lk/debugger_agent_test.go @@ -19,11 +19,13 @@ import ( "context" "encoding/json" "net" + "sync/atomic" "testing" "time" "github.com/stretchr/testify/require" + "github.com/livekit/livekit-cli/v2/pkg/console" "github.com/livekit/livekit-cli/v2/pkg/ipc" agent "github.com/livekit/protocol/livekit/agent" @@ -32,16 +34,40 @@ import ( // fakeAgent stands in for the Python/Node agent on the far end of the console // IPC socket: it records requests and lets the test script events/responses. type fakeAgent struct { - t *testing.T - conn net.Conn - reqs chan *agent.SessionRequest - acks chan struct{} // AudioPlaybackFinished acks from the CLI + t *testing.T + conn net.Conn + reqs chan *agent.SessionRequest + acks chan struct{} // AudioPlaybackFinished acks from the CLI + spoken chan string // text the session asked TTS to speak + heard atomic.Int64 // speech samples (non-silence) received on the mic input } +// fakeSpeech is what the fake TTS produces for any text: 100ms of a constant, +// non-silent sample, so the mic input can be told apart from silence. +const fakeSpeechSamples = 4800 + +// newFakeAgent connects a text-mode session to a fake agent. func newFakeAgent(t *testing.T) (*fakeAgent, *agentSession) { + t.Helper() + return newFakeAgentMode(t, false) +} + +// newAudioFakeAgent connects an audio-mode session to a fake agent, with a +// fake TTS that records what it was asked to speak. +func newAudioFakeAgent(t *testing.T) (*fakeAgent, *agentSession) { + t.Helper() + return newFakeAgentMode(t, true) +} + +func newFakeAgentMode(t *testing.T, audio bool) (*fakeAgent, *agentSession) { t.Helper() client, server := net.Pipe() - fa := &fakeAgent{t: t, conn: client, reqs: make(chan *agent.SessionRequest, 16), acks: make(chan struct{}, 16)} + fa := &fakeAgent{ + t: t, conn: client, + reqs: make(chan *agent.SessionRequest, 16), + acks: make(chan struct{}, 16), + spoken: make(chan string, 16), + } go func() { for { msg := &agent.AgentSessionMessage{} @@ -54,10 +80,28 @@ func newFakeAgent(t *testing.T) (*fakeAgent, *agentSession) { fa.reqs <- m.Request case *agent.AgentSessionMessage_AudioPlaybackFinished: fa.acks <- struct{}{} + case *agent.AgentSessionMessage_AudioInput: + for _, v := range console.BytesToSamples(m.AudioInput.Data) { + if v != 0 { + fa.heard.Add(1) + } + } } } }() - sess := newAgentSession(server, nil) + var speak speakFunc = func(ctx context.Context, text string, onAudio func([]int16)) error { + fa.spoken <- text + speech := make([]int16, fakeSpeechSamples) + for i := range speech { + speech[i] = 1000 + } + onAudio(speech) + return nil + } + if !audio { + speak = nil + } + sess := newAgentSession(server, nil, speak) t.Cleanup(func() { client.Close(); server.Close() }) return fa, sess } @@ -82,6 +126,29 @@ func (fa *fakeAgent) sendAssistant(text string) { }}) } +func (fa *fakeAgent) sendUser(text string) { + fa.sendEvent(&agent.AgentSessionEvent{Event: &agent.AgentSessionEvent_ConversationItemAdded_{ + ConversationItemAdded: &agent.AgentSessionEvent_ConversationItemAdded{Item: userItem(text)}, + }}) +} + +func (fa *fakeAgent) sendError(msg string) { + fa.sendEvent(&agent.AgentSessionEvent{Event: &agent.AgentSessionEvent_Error_{ + Error: &agent.AgentSessionEvent_Error{Message: msg}, + }}) +} + +// nextSpoken waits for the session to speak a user turn. +func (fa *fakeAgent) nextSpoken() string { + select { + case text := <-fa.spoken: + return text + case <-time.After(5 * time.Second): + fa.t.Fatal("timed out waiting for the CLI to speak a turn") + return "" + } +} + func (fa *fakeAgent) respond(resp *agent.SessionResponse) { fa.send(&agent.AgentSessionMessage{Message: &agent.AgentSessionMessage_Response{Response: resp}}) } @@ -455,3 +522,231 @@ func TestObserveStreamDoesNotAffectTurns(t *testing.T) { require.Len(t, sess.RecentEvents(1), 1) require.Equal(t, "unprompted", sess.RecentEvents(1)[0].Text) } + +// sayAsync runs Say in the background, collecting what it streams. +type sayRun struct { + done chan struct{} + got []turnEvent + res turnResult + err error +} + +func sayAsync(ctx context.Context, sess *agentSession, text string) *sayRun { + r := &sayRun{done: make(chan struct{})} + go func() { + defer close(r.done) + r.res, r.err = sess.Say(ctx, text, func(e turnEvent) { r.got = append(r.got, e) }) + }() + return r +} + +func (r *sayRun) wait(t *testing.T) { + t.Helper() + select { + case <-r.done: + case <-time.After(10 * time.Second): + t.Fatal("Say did not return") + } +} + +func TestAudioSayStreamsEventsAndEarlierItems(t *testing.T) { + fa, sess := newAudioFakeAgent(t) + + // A greeting nobody was listening for must be held, not dropped. + fa.sendAssistant("Welcome!") + require.Eventually(t, func() bool { + sess.mu.Lock() + defer sess.mu.Unlock() + return len(sess.undelivered) == 1 + }, time.Second, 10*time.Millisecond) + + run := sayAsync(context.Background(), sess, "My name is Sara, S A R A") + require.Equal(t, "My name is Sara, S A R A", fa.nextSpoken()) + require.True(t, sess.TurnInProgress()) + + // The agent hears the speech (imperfectly), calls a tool, and replies. + fa.sendUser("My name is Sarah. S A R A.") + fa.sendState(agent.AgentState_AS_THINKING) + fa.sendEvent(&agent.AgentSessionEvent{Event: &agent.AgentSessionEvent_FunctionToolsExecuted_{ + FunctionToolsExecuted: &agent.AgentSessionEvent_FunctionToolsExecuted{ + FunctionCalls: []*agent.FunctionCall{{CallId: "c1", Name: "lookup", Arguments: `{"q":1}`}}, + FunctionCallOutputs: []*agent.FunctionCallOutput{{CallId: "c1", Output: "42"}}, + }, + }}) + fa.sendState(agent.AgentState_AS_SPEAKING) + fa.sendAssistant("The answer is 42.") + fa.sendState(agent.AgentState_AS_LISTENING) + + run.wait(t) + require.NoError(t, run.err) + require.False(t, sess.TurnInProgress()) + require.Equal(t, "The answer is 42.", run.res.Reply) + require.Equal(t, "My name is Sarah. S A R A.", run.res.Heard) + require.False(t, run.res.Silent) + require.EqualValues(t, fakeSpeechSamples, fa.heard.Load(), "the whole utterance must reach the agent's mic input") + + types := make([]string, 0, len(run.got)) + for _, e := range run.got { + types = append(types, e.Type+"/"+e.Role) + } + require.Equal(t, []string{"message/assistant", "message/user", "tool_call/", "message/assistant"}, types) + require.True(t, run.got[0].Earlier, "greeting should be flagged as predating the turn") + require.Equal(t, "Welcome!", run.got[0].Text) + require.Equal(t, "lookup", run.got[2].Name) + require.Equal(t, `{"q":1}`, run.got[2].Arguments) + require.Equal(t, "42", run.got[2].Output) + require.Equal(t, 1, sess.Turns()) +} + +func TestAudioSayReportsAgentError(t *testing.T) { + fa, sess := newAudioFakeAgent(t) + run := sayAsync(context.Background(), sess, "hi") + fa.nextSpoken() + fa.sendUser("hi") + fa.sendState(agent.AgentState_AS_THINKING) + fa.sendError("llm exploded") + fa.sendState(agent.AgentState_AS_LISTENING) + run.wait(t) + require.ErrorContains(t, run.err, "llm exploded") +} + +func TestAudioSayReportsSilence(t *testing.T) { + fa, sess := newAudioFakeAgent(t) + sess.replyWait = 200 * time.Millisecond + run := sayAsync(context.Background(), sess, "hello?") + fa.nextSpoken() + run.wait(t) + require.NoError(t, run.err) + require.True(t, run.res.Silent) + require.Empty(t, run.res.Heard) +} + +func TestAudioSayWaitsForTheAgentToHearTheUser(t *testing.T) { + fa, sess := newAudioFakeAgent(t) + run := sayAsync(context.Background(), sess, "My name is Sara") + fa.nextSpoken() + + // The tail of an earlier reply lands while the user is speaking; it must + // not pass for the answer to this turn. + fa.sendState(agent.AgentState_AS_SPEAKING) + fa.sendAssistant("June tenth has already passed.") + fa.sendState(agent.AgentState_AS_LISTENING) + time.Sleep(turnSettle + 500*time.Millisecond) + select { + case <-run.done: + t.Fatal("the turn ended before the agent heard the user") + default: + } + + fa.sendUser("My name is Sarah") + fa.sendState(agent.AgentState_AS_SPEAKING) + fa.sendAssistant("Thanks, Sarah.") + fa.sendState(agent.AgentState_AS_LISTENING) + run.wait(t) + require.NoError(t, run.err) + require.False(t, run.res.Silent) + require.Equal(t, "My name is Sarah", run.res.Heard) + require.Equal(t, "June tenth has already passed.\nThanks, Sarah.", run.res.Reply) +} + +func TestAudioSayReportsSilenceAfterHearing(t *testing.T) { + // The agent hears the user but never answers: the failure to catch. + fa, sess := newAudioFakeAgent(t) + sess.replyWait = 300 * time.Millisecond + run := sayAsync(context.Background(), sess, "Are you there?") + fa.nextSpoken() + fa.sendUser("Are you there?") + run.wait(t) + require.NoError(t, run.err) + require.True(t, run.res.Silent) + require.Equal(t, "Are you there?", run.res.Heard) +} + +func TestAudioSayTimeoutLeavesLateEventsForNextTurn(t *testing.T) { + fa, sess := newAudioFakeAgent(t) + + ctx, cancel := context.WithTimeout(context.Background(), 400*time.Millisecond) + defer cancel() + run := sayAsync(ctx, sess, "slow") + fa.nextSpoken() + fa.sendState(agent.AgentState_AS_THINKING) + run.wait(t) + require.ErrorContains(t, run.err, "timed out") + require.False(t, sess.TurnInProgress()) + + // The agent finishes late: its output is held for the next turn. + fa.sendAssistant("late reply") + require.Eventually(t, func() bool { + sess.mu.Lock() + defer sess.mu.Unlock() + return len(sess.undelivered) == 1 + }, time.Second, 10*time.Millisecond) + earlier := sess.takeUndelivered() + require.Len(t, earlier, 1) + require.True(t, earlier[0].Earlier) +} + +// sendAgentAudio sends d of agent speech followed by a playback flush. +func (fa *fakeAgent) sendAgentAudio(d time.Duration) { + fa.send(&agent.AgentSessionMessage{Message: &agent.AgentSessionMessage_AudioOutput{ + AudioOutput: &agent.AgentSessionMessage_ConsoleIO_AudioFrame{ + SampleRate: 48000, NumChannels: 1, SamplesPerChannel: uint32(48000 * d / time.Second), + }, + }}) + fa.send(&agent.AgentSessionMessage{Message: &agent.AgentSessionMessage_AudioPlaybackFlush{ + AudioPlaybackFlush: &agent.AgentSessionMessage_ConsoleIO_AudioPlaybackFlush{}, + }}) +} + +func TestAudioAcksPlaybackAfterPlayout(t *testing.T) { + fa, _ := newAudioFakeAgent(t) + start := time.Now() + fa.sendAgentAudio(400 * time.Millisecond) + select { + case <-fa.acks: + require.GreaterOrEqual(t, time.Since(start), 350*time.Millisecond, "flush acked before the audio could have played") + case <-time.After(2 * time.Second): + t.Fatal("no AudioPlaybackFinished ack for the flush") + } +} + +func TestAudioClearCancelsPlaybackAck(t *testing.T) { + fa, _ := newAudioFakeAgent(t) + fa.sendAgentAudio(300 * time.Millisecond) + fa.send(&agent.AgentSessionMessage{Message: &agent.AgentSessionMessage_AudioPlaybackClear{ + AudioPlaybackClear: &agent.AgentSessionMessage_ConsoleIO_AudioPlaybackClear{}, + }}) + select { + case <-fa.acks: + t.Fatal("an interrupted playout must not be acked as finished") + case <-time.After(600 * time.Millisecond): + } +} + +func TestSameWords(t *testing.T) { + require.True(t, sameWords("My name is Sara, S A R A", "my name is sara. S A R A!")) + require.False(t, sameWords("My name is Sara", "My name is Sarah")) +} + +func TestAudioSayWaitsForHandoffIntroduction(t *testing.T) { + fa, sess := newAudioFakeAgent(t) + run := sayAsync(context.Background(), sess, "billing please") + fa.nextSpoken() + fa.sendUser("billing please") + fa.sendState(agent.AgentState_AS_THINKING) + old := "assistant" + fa.sendEvent(&agent.AgentSessionEvent{Event: &agent.AgentSessionEvent_ConversationItemAdded_{ + ConversationItemAdded: &agent.AgentSessionEvent_ConversationItemAdded{Item: &agent.ChatContext_ChatItem{ + Item: &agent.ChatContext_ChatItem_AgentHandoff{AgentHandoff: &agent.AgentHandoff{OldAgentId: &old, NewAgentId: "billing"}}, + }}, + }}) + // The old agent goes quiet before the new one has introduced itself. + fa.sendState(agent.AgentState_AS_LISTENING) + time.Sleep(turnSettle + 500*time.Millisecond) + fa.sendAssistant("Hi, billing here.") + + run.wait(t) + require.Len(t, run.got, 3) + require.Equal(t, "handoff", run.got[1].Type) + require.Equal(t, "Hi, billing here.", run.got[2].Text) +} diff --git a/cmd/lk/debugger_audio.go b/cmd/lk/debugger_audio.go new file mode 100644 index 00000000..bceb6d8e --- /dev/null +++ b/cmd/lk/debugger_audio.go @@ -0,0 +1,253 @@ +// Copyright 2025 LiveKit, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package main + +import ( + "context" + "encoding/base64" + "encoding/binary" + "encoding/json" + "fmt" + "net/http" + "net/url" + "strconv" + "strings" + "sync" + "time" + + "github.com/gorilla/websocket" + + "github.com/livekit/protocol/auth" + + "github.com/livekit/livekit-cli/v2/pkg/console" +) + +// The console wire carries 48 kHz mono PCM both ways. +const ( + micFrameSamples = console.SampleRate / 50 // 20ms + micFrameTime = 20 * time.Millisecond +) + +// micStream stands in for the user's microphone: the agent receives a +// continuous real-time stream, speech when some is queued and silence +// otherwise, so its VAD and endpointing see the same input as from a live mic. +type micStream struct { + mu sync.Mutex + pending []int16 + waiters []chan struct{} +} + +// push queues speech to be played after anything already queued. +func (m *micStream) push(samples []int16) { + m.mu.Lock() + m.pending = append(m.pending, samples...) + m.mu.Unlock() +} + +// drained returns a channel closed once everything queued so far has been sent. +func (m *micStream) drained() <-chan struct{} { + ch := make(chan struct{}) + m.mu.Lock() + defer m.mu.Unlock() + if len(m.pending) == 0 { + close(ch) + } else { + m.waiters = append(m.waiters, ch) + } + return ch +} + +// run sends one frame per micFrameTime until send fails or done closes. +func (m *micStream) run(done <-chan struct{}, send func([]int16) error) { + next := time.Now() + for { + frame := make([]int16, micFrameSamples) + m.mu.Lock() + n := copy(frame, m.pending) + m.pending = m.pending[n:] + var woken []chan struct{} + if len(m.pending) == 0 { + woken, m.waiters = m.waiters, nil + } + m.mu.Unlock() + + if err := send(frame); err != nil { + return + } + for _, ch := range woken { + close(ch) + } + + next = next.Add(micFrameTime) + wait := time.Until(next) + if wait < -micFrameTime { + // Fell behind; resync rather than bursting to catch up, which would + // compress the audio the agent hears. + next = time.Now() + wait = 0 + } + select { + case <-done: + return + case <-time.After(wait): + } + } +} + +// playout stands in for the user's speaker: agent audio "plays" in real time +// with nothing audible, so a flush is acknowledged only once the audio before +// it would have finished, as a real speaker would. +type playout struct { + mu sync.Mutex + end time.Time // when the buffered agent audio finishes playing + cancel chan struct{} // closed to abandon the pending flush acknowledgement +} + +func (p *playout) write(samples, sampleRate uint32) { + if sampleRate == 0 { + return + } + d := time.Duration(samples) * time.Second / time.Duration(sampleRate) + p.mu.Lock() + if now := time.Now(); p.end.Before(now) { + p.end = now + } + p.end = p.end.Add(d) + p.mu.Unlock() +} + +// flush calls ack once the buffered audio has played, unless a clear or a +// newer flush comes first. +func (p *playout) flush(ack func()) { + p.mu.Lock() + if p.cancel != nil { + close(p.cancel) + } + cancel := make(chan struct{}) + p.cancel = cancel + wait := time.Until(p.end) + p.mu.Unlock() + + go func() { + select { + case <-time.After(wait): + ack() + case <-cancel: + } + }() +} + +// clear drops the buffered audio; the agent accounts for the interruption itself. +func (p *playout) clear() { + p.mu.Lock() + p.end = time.Time{} + if p.cancel != nil { + close(p.cancel) + p.cancel = nil + } + p.mu.Unlock() +} + +// speakFunc synthesizes text as 48 kHz mono PCM, handing chunks to onAudio as +// they arrive. +type speakFunc func(ctx context.Context, text string, onAudio func([]int16)) error + +const ( + userTTSModel = "cartesia/sonic-3" + inferenceURL = "https://agent-gateway.livekit.cloud/v1" + inferenceStageURL = "https://agent-gateway.staging.livekit.cloud/v1" +) + +// inferenceTTS speaks the user's lines through LiveKit Inference, using the +// same gateway protocol as the agents SDKs' inference.TTS. +type inferenceTTS struct { + livekitURL, apiKey, apiSecret string +} + +func (t inferenceTTS) gatewayURL() string { + base := inferenceURL + if strings.Contains(t.livekitURL, ".staging.livekit.cloud") { + base = inferenceStageURL + } + return strings.Replace(base, "http", "ws", 1) + "/tts?" + url.Values{"model": {userTTSModel}}.Encode() +} + +func (t inferenceTTS) speak(ctx context.Context, text string, onAudio func([]int16)) error { + token, err := auth.NewAccessToken(t.apiKey, t.apiSecret). + SetIdentity("lk-agent-debugger"). + SetInferenceGrant(&auth.InferenceGrant{Perform: true}). + SetValidFor(10 * time.Minute). + ToJWT() + if err != nil { + return fmt.Errorf("text-to-speech: %w", err) + } + ws, resp, err := websocket.DefaultDialer.DialContext(ctx, t.gatewayURL(), http.Header{"Authorization": {"Bearer " + token}}) + if err != nil { + if resp != nil { + return fmt.Errorf("text-to-speech: LiveKit Inference returned %s", resp.Status) + } + return fmt.Errorf("text-to-speech: %w", err) + } + defer ws.Close() + stop := context.AfterFunc(ctx, func() { ws.Close() }) + defer stop() + + for _, msg := range []map[string]any{ + {"type": "session.create", "sample_rate": strconv.Itoa(console.SampleRate), "encoding": "pcm_s16le", "model": userTTSModel, "extra": map[string]any{}}, + {"type": "input_transcript", "transcript": text + " ", "generation_config": map[string]any{"model": userTTSModel}, "extra": map[string]any{}}, + {"type": "session.flush"}, + } { + if err := ws.WriteJSON(msg); err != nil { + return fmt.Errorf("text-to-speech: %w", err) + } + } + + var odd []byte // a sample split across chunks + for { + _, data, err := ws.ReadMessage() + if err != nil { + if ctx.Err() != nil { + return ctx.Err() + } + return fmt.Errorf("text-to-speech: %w", err) + } + var msg struct { + Type string `json:"type"` + Audio string `json:"audio"` + } + if err := json.Unmarshal(data, &msg); err != nil { + continue + } + switch msg.Type { + case "output_audio": + pcm, err := base64.StdEncoding.DecodeString(msg.Audio) + if err != nil { + return fmt.Errorf("text-to-speech: %w", err) + } + pcm = append(odd, pcm...) + n := len(pcm) / 2 + samples := make([]int16, n) + for i := range samples { + samples[i] = int16(binary.LittleEndian.Uint16(pcm[2*i:])) + } + odd = append([]byte(nil), pcm[2*n:]...) + onAudio(samples) + case "done": + return nil + case "error": + return fmt.Errorf("text-to-speech: LiveKit Inference returned an error: %s", data) + } + } +} diff --git a/cmd/lk/debugger_daemon.go b/cmd/lk/debugger_daemon.go index 07fcec1d..f1eed08f 100644 --- a/cmd/lk/debugger_daemon.go +++ b/cmd/lk/debugger_daemon.go @@ -89,6 +89,12 @@ func runSessionDaemon() { entrypoint: entry, projectType: ptype, startedAt: time.Now(), + audio: os.Getenv(envSessionAudio) != "", + tts: inferenceTTS{ + livekitURL: os.Getenv(envSessionURL), + apiKey: os.Getenv(envSessionAPIKey), + apiSecret: os.Getenv(envSessionAPISecret), + }, idleTimeout: idleTimeout, agentReady: make(chan struct{}), logSubs: make(map[chan string]struct{}), @@ -123,9 +129,11 @@ func runSessionDaemon() { os.Exit(1) } - ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) - _ = d.session.SetTextMode(ctx) - cancel() + if !d.audio { + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + _ = d.session.SetTextMode(ctx) + cancel() + } // Let an agent that greets first finish, so `start` can show the greeting. d.session.WaitForGreeting(context.Background(), greetingSettle, greetingMaxWait) signalReady(ready, "ready") @@ -184,6 +192,8 @@ type sessionDaemon struct { startedAt time.Time idleTimeout time.Duration lastActive atomic.Int64 // unix nanos of the last control command + audio bool // turns are spoken instead of sent as text + tts inferenceTTS sessionMu sync.Mutex session *agentSession @@ -262,7 +272,11 @@ func (d *sessionDaemon) handleConn(conn net.Conn) { conn.Close() return } - d.session = newAgentSession(conn, reader) + var speak speakFunc + if d.audio { + speak = d.tts.speak + } + d.session = newAgentSession(conn, reader, speak) d.sessionMu.Unlock() close(d.agentReady) } @@ -421,6 +435,7 @@ func (d *sessionDaemon) handleSay(conn *controlConn, req controlRequest) { done := controlReply{ Done: true, Reply: res.Reply, + Heard: res.Heard, Silent: res.Silent, DurationMs: res.Duration.Milliseconds(), } @@ -560,6 +575,7 @@ type sessionStatus struct { ProjectDir string `json:"project_dir"` Entrypoint string `json:"entrypoint"` ProjectType string `json:"project_type"` + Audio bool `json:"audio"` StartedAt time.Time `json:"started_at"` UptimeSeconds int64 `json:"uptime_seconds"` Turns int `json:"turns"` @@ -582,6 +598,7 @@ func (d *sessionDaemon) collectStatus() *sessionStatus { ProjectDir: d.projectDir, Entrypoint: d.entrypoint, ProjectType: string(d.projectType), + Audio: d.audio, StartedAt: d.startedAt, UptimeSeconds: int64(time.Since(d.startedAt).Seconds()), Turns: d.session.Turns(), diff --git a/cmd/lk/debugger_e2e_test.go b/cmd/lk/debugger_e2e_test.go index 7a152d63..3b2dbc0e 100644 --- a/cmd/lk/debugger_e2e_test.go +++ b/cmd/lk/debugger_e2e_test.go @@ -17,6 +17,7 @@ package main import ( "bytes" "context" + "encoding/json" "net" "os" "os/exec" @@ -32,8 +33,9 @@ import ( const sessionE2ETimeout = 5 * time.Second // TestSessionE2E drives the real `lk agent debugger` lifecycle end to end: -// build the binary, `start` the detached daemon, `say` to make the model echo -// a token (asserting the CLI→daemon→agent→LLM round-trip), `stop`, confirm a +// build the binary, then in text mode and in audio mode: `start` the detached +// daemon, `say` a line the model echoes (asserting the CLI→daemon→agent→LLM +// round-trip, through TTS and the agent's STT in audio mode), `stop`, confirm a // second `say` cannot still reach the agent, then confirm the daemon exited // (nothing answers on the port). // @@ -59,104 +61,116 @@ func TestSessionE2E(t *testing.T) { bin := buildLK(t) - type runResult struct { - stdout string - stderr string - exitCode int + for _, audio := range []bool{false, true} { + t.Run(sessionMode(audio), func(t *testing.T) { + startArgs := []string{"agent", "debugger", "start", "--port", port} + if audio { + startArgs = append(startArgs, "--audio") + } + + type runResult struct { + stdout string + stderr string + exitCode int + } + + runCapture := func(timeout time.Duration, args ...string) (runResult, error) { + ctx, cancel := context.WithTimeout(context.Background(), timeout) + defer cancel() + cmd := exec.CommandContext(ctx, bin, args...) + cmd.Env = os.Environ() + var stdout, stderr bytes.Buffer + cmd.Stdout = &stdout + cmd.Stderr = &stderr + err := cmd.Run() + require.NotNil(t, cmd.ProcessState, "command did not start: %v", err) + + return runResult{ + stdout: stdout.String(), + stderr: stderr.String(), + exitCode: cmd.ProcessState.ExitCode(), + }, err + } + + run := func(timeout time.Duration, args ...string) (string, error) { + res, err := runCapture(timeout, args...) + return res.stdout + res.stderr, err + } + + portIsFree := func() bool { + conn, derr := net.DialTimeout("tcp", "127.0.0.1:"+port, 200*time.Millisecond) + if derr != nil { + return true // refused -> daemon exited + } + conn.Close() + return false + } + + // Best-effort teardown so a mid-run failure doesn't leave the daemon alive. + t.Cleanup(func() { + _, _ = run(sessionE2ETimeout, "agent", "debugger", "stop", "--port", port) + }) + + // start: launches the detached daemon and returns once the agent is ready. + // The CLI itself waits up to 65s for daemon readiness (awaitDaemonReady), + // so give it just over that: on a cold Windows runner the chain of first + // execs (lk.exe, uv, python) can alone eat a tighter budget, and killing + // the command mid-wait loses the CLI's own error report. + startOut, err := run(70*time.Second, append(startArgs, entrypoint)...) + require.NoError(t, err, "session start failed:\n%s", startOut) + require.Contains(t, startOut, "Session started in "+sessionMode(audio)+" mode.", "start did not report readiness:\n%s", startOut) + + // say: the token must reach the agent (as text, or through its STT in + // audio mode) and come back in the reply. + token := "pineapple" + sayRes, err := runCapture(60*time.Second, "agent", "debugger", "say", "--port", port, "--json", + "Repeat this word back to me: "+token) + sayOut := sayRes.stdout + sayRes.stderr + require.NoError(t, err, "session say failed:\n%s", sayOut) + var turn sayJSON + require.NoError(t, json.Unmarshal([]byte(sayRes.stdout), &turn), "say --json output:\n%s", sayOut) + require.Containsf(t, strings.ToLower(turn.Heard), token, "agent did not hear the token; say output:\n%s", sayOut) + require.Containsf(t, strings.ToLower(turn.Reply), token, "agent did not echo the token back; say output:\n%s", sayOut) + + // history: the agent's own record of the conversation must contain the turn. + histOut, err := run(sessionE2ETimeout, "agent", "debugger", "chat-history", "--port", port) + require.NoError(t, err, "session history failed:\n%s", histOut) + require.GreaterOrEqualf(t, strings.Count(strings.ToLower(histOut), token), 2, + "history did not contain both the prompt and the reply:\n%s", histOut) + + // status/logs: both must answer while the session is up. + statusOut, err := run(sessionE2ETimeout, "agent", "debugger", "status", "--port", port) + require.NoError(t, err, "session status failed:\n%s", statusOut) + require.Contains(t, statusOut, "Session running", "status did not report a running session:\n%s", statusOut) + logsOut, err := run(sessionE2ETimeout, "agent", "debugger", "logs", "-n", "5", "--port", port) + require.NoError(t, err, "session logs failed:\n%s", logsOut) + + stopOut, err := run(sessionE2ETimeout, "agent", "debugger", "stop", "--port", port) + require.NoError(t, err, "session stop failed:\n%s", stopOut) + require.Contains(t, stopOut, "Session ended.", "stop did not confirm shutdown:\n%s", stopOut) + + require.Eventually(t, portIsFree, sessionE2ETimeout, 200*time.Millisecond, + "session daemon still listening on port %s after stop", port) + + // After a successful match and shutdown, another say must not reach a live + // agent or reproduce the token. + afterStopSay, err := runCapture(sessionE2ETimeout, "agent", "debugger", "say", "--port", port, + "Repeat this word back to me: "+token) + afterStopSayOut := afterStopSay.stdout + afterStopSay.stderr + require.Error(t, err, "session say unexpectedly succeeded after stop:\n%s", afterStopSayOut) + require.Equal(t, 1, afterStopSay.exitCode, + "session say after stop exited with wrong code; stdout:\n%s\nstderr:\n%s", + afterStopSay.stdout, afterStopSay.stderr) + require.Truef(t, strings.HasPrefix(afterStopSayOut, "no session running"), + "session say after stop output did not start with no session running; stdout:\n%s\nstderr:\n%s", + afterStopSay.stdout, afterStopSay.stderr) + require.NotContains(t, afterStopSayOut, token, + "session say after stop unexpectedly contained the matched token; stdout:\n%s\nstderr:\n%s", + afterStopSay.stdout, afterStopSay.stderr) + + require.True(t, portIsFree(), "session daemon started listening again on port %s after failed say", port) + }) } - - runCapture := func(timeout time.Duration, args ...string) (runResult, error) { - ctx, cancel := context.WithTimeout(context.Background(), timeout) - defer cancel() - cmd := exec.CommandContext(ctx, bin, args...) - cmd.Env = os.Environ() - var stdout, stderr bytes.Buffer - cmd.Stdout = &stdout - cmd.Stderr = &stderr - err := cmd.Run() - require.NotNil(t, cmd.ProcessState, "command did not start: %v", err) - - return runResult{ - stdout: stdout.String(), - stderr: stderr.String(), - exitCode: cmd.ProcessState.ExitCode(), - }, err - } - - run := func(timeout time.Duration, args ...string) (string, error) { - res, err := runCapture(timeout, args...) - return res.stdout + res.stderr, err - } - - portIsFree := func() bool { - conn, derr := net.DialTimeout("tcp", "127.0.0.1:"+port, 200*time.Millisecond) - if derr != nil { - return true // refused -> daemon exited - } - conn.Close() - return false - } - - // Best-effort teardown so a mid-run failure doesn't leave the daemon alive. - t.Cleanup(func() { - _, _ = run(sessionE2ETimeout, "agent", "debugger", "stop", "--port", port) - }) - - // start: launches the detached daemon and returns once the agent is ready. - // The CLI itself waits up to 65s for daemon readiness (awaitDaemonReady), - // so give it just over that: on a cold Windows runner the chain of first - // execs (lk.exe, uv, python) can alone eat a tighter budget, and killing - // the command mid-wait loses the CLI's own error report. - startOut, err := run(70*time.Second, "agent", "debugger", "start", "--port", port, entrypoint) - require.NoError(t, err, "session start failed:\n%s", startOut) - require.Contains(t, startOut, "Session started.", "start did not report readiness:\n%s", startOut) - - // say: the token appears once in the echoed prompt and again in the reply, so - // >=2 occurrences proves the agent answered, not just the local echo. - token := "PINEAPPLE7351" - sayOut, err := run(sessionE2ETimeout, "agent", "debugger", "say", "--port", port, - "Repeat this token back to me exactly and nothing else: "+token) - require.NoError(t, err, "session say failed:\n%s", sayOut) - require.GreaterOrEqualf(t, strings.Count(sayOut, token), 2, - "agent did not echo the token back; say output:\n%s", sayOut) - - // history: the agent's own record of the conversation must contain the turn. - histOut, err := run(sessionE2ETimeout, "agent", "debugger", "chat-history", "--port", port) - require.NoError(t, err, "session history failed:\n%s", histOut) - require.GreaterOrEqualf(t, strings.Count(histOut, token), 2, - "history did not contain both the prompt and the reply:\n%s", histOut) - - // status/logs: both must answer while the session is up. - statusOut, err := run(sessionE2ETimeout, "agent", "debugger", "status", "--port", port) - require.NoError(t, err, "session status failed:\n%s", statusOut) - require.Contains(t, statusOut, "Session running", "status did not report a running session:\n%s", statusOut) - logsOut, err := run(sessionE2ETimeout, "agent", "debugger", "logs", "-n", "5", "--port", port) - require.NoError(t, err, "session logs failed:\n%s", logsOut) - - stopOut, err := run(sessionE2ETimeout, "agent", "debugger", "stop", "--port", port) - require.NoError(t, err, "session stop failed:\n%s", stopOut) - require.Contains(t, stopOut, "Session ended.", "stop did not confirm shutdown:\n%s", stopOut) - - require.Eventually(t, portIsFree, sessionE2ETimeout, 200*time.Millisecond, - "session daemon still listening on port %s after stop", port) - - // After a successful match and shutdown, another say must not reach a live - // agent or reproduce the token. - afterStopSay, err := runCapture(sessionE2ETimeout, "agent", "debugger", "say", "--port", port, - "Repeat this token back to me exactly and nothing else: "+token) - afterStopSayOut := afterStopSay.stdout + afterStopSay.stderr - require.Error(t, err, "session say unexpectedly succeeded after stop:\n%s", afterStopSayOut) - require.Equal(t, 1, afterStopSay.exitCode, - "session say after stop exited with wrong code; stdout:\n%s\nstderr:\n%s", - afterStopSay.stdout, afterStopSay.stderr) - require.Truef(t, strings.HasPrefix(afterStopSayOut, "no session running"), - "session say after stop output did not start with no session running; stdout:\n%s\nstderr:\n%s", - afterStopSay.stdout, afterStopSay.stderr) - require.NotContains(t, afterStopSayOut, token, - "session say after stop unexpectedly contained the matched token; stdout:\n%s\nstderr:\n%s", - afterStopSay.stdout, afterStopSay.stderr) - - require.True(t, portIsFree(), "session daemon started listening again on port %s after failed say", port) } // buildLK returns the path to the lk binary under test. If LK_SESSION_E2E_BIN diff --git a/cmd/lk/testdata/echo-agent-node/agent.ts b/cmd/lk/testdata/echo-agent-node/agent.ts index 85c72d99..52998e61 100644 --- a/cmd/lk/testdata/echo-agent-node/agent.ts +++ b/cmd/lk/testdata/echo-agent-node/agent.ts @@ -2,8 +2,9 @@ * Minimal one-file echo agent for the `lk agent debugger` e2e test -- the Node * sibling of testdata/echo-agent/agent.py. * - * Driven in text mode, so an LLM is the only component needed. Echoes the - * user's text verbatim, which the test asserts on. + * Has the full voice pipeline (STT, LLM, TTS) through LiveKit Inference, so + * the test can drive it in both text and audio mode. Echoes what the user + * says, which the test asserts on. */ import { type JobContext, ServerOptions, cli, defineAgent, inference, voice } from '@livekit/agents'; import 'dotenv/config'; @@ -12,16 +13,16 @@ import { fileURLToPath } from 'node:url'; export default defineAgent({ entry: async (ctx: JobContext) => { const session = new voice.AgentSession({ + stt: new inference.STT({ model: 'deepgram/nova-3' }), llm: new inference.LLM({ model: 'openai/gpt-4o-mini' }), + tts: new inference.TTS({ model: 'cartesia/sonic-3' }), }); await session.start({ agent: new voice.Agent({ instructions: - 'You are an echo bot. Reply with exactly the text the user sends, verbatim, and nothing else.', + 'You are an echo bot. Reply with exactly what the user says, verbatim, and nothing else.', }), room: ctx.room, - // No TTS, so disable audio output or the turn crashes in the tts node. - outputOptions: { audioEnabled: false }, }); await ctx.connect(); }, diff --git a/cmd/lk/testdata/echo-agent/agent.py b/cmd/lk/testdata/echo-agent/agent.py index 5f6ed828..76ac5c28 100644 --- a/cmd/lk/testdata/echo-agent/agent.py +++ b/cmd/lk/testdata/echo-agent/agent.py @@ -1,7 +1,8 @@ """Minimal one-file echo agent for the `lk agent debugger` e2e test. -Driven in text mode, so an LLM is the only component needed. Echoes the user's -text verbatim, which the test asserts on. +Has the full voice pipeline (STT, LLM, TTS) through LiveKit Inference, so the +test can drive it in both text and audio mode. Echoes what the user says, which +the test asserts on. """ from dotenv import load_dotenv @@ -14,18 +15,20 @@ @server.rtc_session() async def entrypoint(ctx: JobContext): - session = AgentSession(llm=inference.LLM(model="openai/gpt-4o-mini")) + session = AgentSession( + stt=inference.STT(model="deepgram/nova-3"), + llm=inference.LLM(model="openai/gpt-4o-mini"), + tts=inference.TTS(model="cartesia/sonic-3"), + ) await session.start( agent=Agent( instructions=( - "You are an echo bot. Reply with exactly the text the user " - "sends, verbatim, and nothing else." + "You are an echo bot. Reply with exactly what the user says, " + "verbatim, and nothing else." ), ), room=ctx.room, ) - # No TTS, so disable audio output or the turn crashes in tts_node. - session.output.set_audio_enabled(False) await ctx.connect() diff --git a/cmd/lk/transcript.go b/cmd/lk/transcript.go index a4b05e96..e34baf99 100644 --- a/cmd/lk/transcript.go +++ b/cmd/lk/transcript.go @@ -28,7 +28,7 @@ import ( // or handoff is represented is made once. // turnEvent is the normalized, transport-agnostic record of one thing that -// happened in a text-mode agent session: a message, a tool call, a handoff, an +// happened in an agent session: a message, a tool call, a handoff, an // agent error, or an agent log line. The daemon ships these to `say`/`history` // clients as JSON, the scripted console prints them locally, and `--json` // emits them verbatim, so the schema doubles as the machine-readable contract @@ -76,6 +76,12 @@ func (e turnEvent) isOutput() bool { return false } +// isAgentOutput reports whether the event is the agent responding, which +// excludes what it transcribed from the user. +func (e turnEvent) isAgentOutput() bool { + return e.isOutput() && (e.Type != "message" || e.Role != "user") +} + // eventTimestamp formats a time the way turnEvent.Time carries it. func eventTimestamp(t time.Time) string { return t.UTC().Format("2006-01-02T15:04:05.000Z07:00") } diff --git a/go.mod b/go.mod index 2ee88fd6..385defad 100644 --- a/go.mod +++ b/go.mod @@ -18,6 +18,7 @@ require ( github.com/go-task/task/v3 v3.51.1 github.com/google/go-containerregistry v0.21.7 github.com/google/go-querystring v1.2.0 + github.com/gorilla/websocket v1.5.4-0.20250319132907-e064f32e3674 github.com/joho/godotenv v1.5.1 github.com/klauspost/compress v1.20.0 github.com/livekit/protocol v1.51.1-0.20260915182407-3c0fcdbd25df @@ -141,7 +142,6 @@ require ( github.com/google/uuid v1.6.0 // indirect github.com/googleapis/enterprise-certificate-proxy v0.3.14 // indirect github.com/googleapis/gax-go/v2 v2.21.0 // indirect - github.com/gorilla/websocket v1.5.4-0.20250319132907-e064f32e3674 // indirect github.com/grpc-ecosystem/grpc-gateway/v2 v2.29.0 // indirect github.com/hashicorp/aws-sdk-go-base/v2 v2.0.0-beta.72 // indirect github.com/hashicorp/go-cleanhttp v0.5.2 // indirect From 7112762e6ba2da0129a3661ae8eff518a4b75fcb Mon Sep 17 00:00:00 2001 From: Jason Lernerman Date: Fri, 25 Sep 2026 15:01:45 -0400 Subject: [PATCH 3/3] docs(agent): note debugger --audio uses LiveKit Inference on the connected Cloud project --- README.md | 2 +- cmd/lk/debugger.go | 7 ++++--- 2 files changed, 5 insertions(+), 4 deletions(-) diff --git a/README.md b/README.md index cf0a2fa8..4a305b11 100644 --- a/README.md +++ b/README.md @@ -483,7 +483,7 @@ Console needs a terminal. If a program rather than a person is going to hold the `lk agent debugger` is built for coding agents such as Claude Code, Codex, and Cursor. When one of them is working on your LiveKit agent, it can use this command to hold a conversation with the agent it just edited, read the replies and tool calls, and decide what to try next, all from the shell, with no human at a microphone. It is an ad-hoc, local stand-in for [Agent Simulations](https://docs.livekit.io/agents/start/testing/simulations/) in which the coding agent plays the user and pays with its own tokens: by default the agent runs in console mode with STT/TTS disabled, so a turn costs only your LLM and tool calls, and nothing touches a LiveKit room. -Start the session with `--audio` to test the audio path instead: each `say` is spoken with LiveKit Inference TTS into the agent's microphone input, and the agent runs its full audio pipeline (STT, turn detection, TTS, or a realtime model that only takes audio), so misheard names, spelled-out digits, and endpointing behave as they do on a call. Each turn then prints what the agent transcribed, with the text you sent alongside when the words differ. The TTS uses your project's credentials, resolved like other `lk` commands. +Start the session with `--audio` to test the audio path instead: each `say` is spoken with LiveKit Inference TTS into the agent's microphone input, and the agent runs its full audio pipeline (STT, turn detection, TTS, or a realtime model that only takes audio), so misheard names, spelled-out digits, and endpointing behave as they do on a call. Each turn then prints what the agent transcribed, with the text you sent alongside when the words differ. The TTS runs on LiveKit Inference and is billed to your connected LiveKit Cloud project, resolved like other `lk` commands. People can use it too, but the output and flags are shaped for a program driving it one command at a time: every turn prints tool calls with their arguments and results, handoffs, errors, and the reply; exit codes are non-zero when a turn fails; `--json` is available everywhere. If you want to talk to your agent yourself, `lk agent console` is the better fit. diff --git a/cmd/lk/debugger.go b/cmd/lk/debugger.go index 94578f18..aa94d13c 100644 --- a/cmd/lk/debugger.go +++ b/cmd/lk/debugger.go @@ -82,7 +82,7 @@ var sessionIdleFlag = &cli.DurationFlag{ var sessionAudioFlag = &cli.BoolFlag{ Name: "audio", - Usage: "Speak each turn into the agent's microphone input instead of sending text, running its full audio pipeline", + Usage: "Speak each turn into the agent's microphone input instead of sending text, running its full audio pipeline (uses LiveKit Inference TTS on your connected LiveKit Cloud project)", } var sessionMetricsFlag = &cli.BoolFlag{ @@ -124,8 +124,9 @@ with --audio to speak each turn instead: the user's line is spoken with LiveKit Inference TTS into the agent's microphone input, and the agent runs its full audio pipeline (STT, turn detection, TTS, or a realtime model that only takes audio), so misheard names, spelled-out digits, and endpointing show up here. -Audio mode uses your project's credentials for the TTS, resolved like other lk -commands (--project, LIVEKIT_* variables, livekit.toml, or the default project). +The TTS runs on LiveKit Inference and is billed to your connected LiveKit Cloud +project, resolved like other lk commands (--project, LIVEKIT_* variables, +livekit.toml, or the default project). This is built for coding agents (Claude Code, Codex, Cursor, ...) and shell scripts: the caller stands in for the user, decides the next line based on the