From 3f46fbf08aad4c7c0374ab3e36d36d06e9e4ad2f Mon Sep 17 00:00:00 2001 From: David Zhao Date: Sat, 27 Jun 2026 20:51:54 +0200 Subject: [PATCH 1/4] display console link after agent start also fixes agent dev assuming it's localhost --- Makefile | 7 +- cmd/lk/agent_dev.go | 239 +++++++++++++++++++++++ cmd/lk/agent_dev_test.go | 206 ++++++++++++++++++++ cmd/lk/agent_reload.go | 99 ---------- cmd/lk/agent_run.go | 174 ++++++++++++++--- cmd/lk/agent_run_test.go | 353 ++++++++++++++++++++++++++++++++++ cmd/lk/agent_watcher.go | 69 ++++--- cmd/lk/simulate_subprocess.go | 5 + go.mod | 16 +- go.sum | 26 ++- 10 files changed, 1007 insertions(+), 187 deletions(-) create mode 100644 cmd/lk/agent_dev.go create mode 100644 cmd/lk/agent_dev_test.go delete mode 100644 cmd/lk/agent_reload.go diff --git a/Makefile b/Makefile index 2d289c338..651af7355 100644 --- a/Makefile +++ b/Makefile @@ -20,10 +20,13 @@ endif # init makes this work from a fresh clone (and under CodeQL, whose checkout may # skip submodules). ALSA headers (libasound2-dev) come from CodeQL's automatic # dependency installation on Linux. -lk$(EXE): +./bin/lk$(EXE): git submodule update --init --recursive CGO_ENABLED=1 go build -o ./bin/lk$(EXE) ./cmd/lk -install: lk$(EXE) +install: ./bin/lk$(EXE) cp ./bin/lk$(EXE) "$(GOBIN)/lk$(EXE)" ln -sf "$(GOBIN)/lk$(EXE)" "$(GOBIN)/livekit-cli$(EXE)" + +clean: + rm -rf ./bin diff --git a/cmd/lk/agent_dev.go b/cmd/lk/agent_dev.go new file mode 100644 index 000000000..bbce670db --- /dev/null +++ b/cmd/lk/agent_dev.go @@ -0,0 +1,239 @@ +package main + +import ( + "errors" + "fmt" + "net" + "sync" + "time" + + agent "github.com/livekit/protocol/livekit/agent" + + "github.com/livekit/livekit-cli/v2/pkg/ipc" +) + +// devServer owns the dev-mode IPC channel between Go and the agent (Python) +// processes. It listens on a loopback port that each (re)started agent connects +// back to; every connection is handled by its own devSession. Reload (capturing +// running jobs from the old process and restoring them in the new one) is one +// feature served over this channel: +// +// 1. Go → old agent: GetRunningJobsRequest → GetRunningJobsResponse (capture) +// 2. New agent → Go: GetRunningJobsRequest → Go replies with the saved jobs (restore) +type devServer struct { + listener *ipc.Listener + mu sync.Mutex + savedJobs *agent.GetRunningAgentJobsResponse + + // onServerInfo, if set, is invoked when a (re)started agent reports its + // ServerInfo over the dev channel (agent name + the LiveKit URL it uses). + onServerInfo func(agentName, url string) +} + +func newDevServer() (*devServer, error) { + ln, err := ipc.Listen("127.0.0.1:0") + if err != nil { + return nil, fmt.Errorf("dev server: %w", err) + } + return &devServer{listener: ln}, nil +} + +func (rs *devServer) addr() string { + return rs.listener.Addr().String() +} + +// newSession wraps a freshly accepted dev-channel connection. The session +// answers inbound GetRunningJobsRequests with the captured jobs (restore) and +// forwards any ServerInfo notifications. +func (rs *devServer) newSession(conn net.Conn) *devSession { + s := newDevSession(conn) + s.onServerInfo = rs.onServerInfo + s.jobsProvider = rs.takeSavedJobs + return s +} + +// takeSavedJobs returns and clears the captured jobs. Jobs are restored to the +// first process that asks; subsequent asks (within the same generation) get none. +func (rs *devServer) takeSavedJobs() *agent.GetRunningAgentJobsResponse { + rs.mu.Lock() + defer rs.mu.Unlock() + saved := rs.savedJobs + rs.savedJobs = nil + return saved +} + +// captureJobs asks the (old) session for its running jobs and stores them so the +// next process can restore them. Best-effort: failures are logged, not fatal. +func (rs *devServer) captureJobs(s *devSession) { + resp, err := s.getRunningJobs(1500 * time.Millisecond) + if err != nil { + out.Warnf("reload: failed to capture running jobs: %v", err) + return + } + if resp != nil { + rs.mu.Lock() + rs.savedJobs = resp + rs.mu.Unlock() + out.Statusf("reload: captured %d running job(s)", len(resp.Jobs)) + } +} + +func (rs *devServer) close() error { + return rs.listener.Close() +} + +// devMsgKind identifies an outbound call awaiting a reply, keyed by the response +// message type. Type-based routing is unambiguous today because the CLI never has +// two outbound calls of the same response type in flight; add a correlation id to +// AgentDevMessage if that ever changes. +type devMsgKind int + +const ( + kindJobsResponse devMsgKind = iota +) + +var ( + errDevSessionClosed = errors.New("dev session closed") + errDevSessionTimeout = errors.New("dev session request timed out") +) + +// devSession owns a single dev-channel IPC connection and multiplexes the +// AgentDevMessage protocol over it. A single read loop (run) dispatches inbound +// notifications (ServerInfo) and peer requests (GetRunningJobsRequest) to their +// handlers, while outbound calls (getRunningJobs) await their matching response. +// This keeps one owner on the connection so the CLI/agent API can grow new +// request/response pairs and callbacks without per-message lifecycle juggling. +type devSession struct { + conn net.Conn + + // onServerInfo, if set, is invoked when the peer pushes a ServerInfo. + onServerInfo func(agentName, url string) + // jobsProvider, if set, supplies the response to an inbound + // GetRunningJobsRequest (the restore side of a reload). + jobsProvider func() *agent.GetRunningAgentJobsResponse + + writeMu sync.Mutex // serializes writes to conn + + mu sync.Mutex + pending map[devMsgKind]chan *agent.AgentDevMessage +} + +func newDevSession(conn net.Conn) *devSession { + return &devSession{ + conn: conn, + pending: make(map[devMsgKind]chan *agent.AgentDevMessage), + } +} + +// run reads and dispatches messages until the connection closes. It is meant to +// run in its own goroutine for the lifetime of the connection. +func (s *devSession) run() { + defer s.failPending() + for { + msg := &agent.AgentDevMessage{} + if err := ipc.ReadProto(s.conn, msg); err != nil { + return + } + switch msg.Message.(type) { + case *agent.AgentDevMessage_ServerInfo: + if s.onServerInfo != nil { + si := msg.GetServerInfo() + s.onServerInfo(si.GetAgentName(), si.GetUrl()) + } + case *agent.AgentDevMessage_GetRunningJobsRequest: + s.handleJobsRequest() + case *agent.AgentDevMessage_GetRunningJobsResponse: + s.deliver(kindJobsResponse, msg) + } + // Unknown message types are ignored, leaving room for protocol growth. + } +} + +// handleJobsRequest answers a peer's GetRunningJobsRequest with the jobs supplied +// by jobsProvider (the restore side of a reload). +func (s *devSession) handleJobsRequest() { + var resp *agent.GetRunningAgentJobsResponse + if s.jobsProvider != nil { + resp = s.jobsProvider() + } + if resp == nil { + resp = &agent.GetRunningAgentJobsResponse{} + } + err := s.write(&agent.AgentDevMessage{ + Message: &agent.AgentDevMessage_GetRunningJobsResponse{GetRunningJobsResponse: resp}, + }) + if err != nil { + out.Warnf("reload: failed to send restore response: %v", err) + } else if len(resp.Jobs) > 0 { + out.Statusf("reload: restored %d job(s) to new process", len(resp.Jobs)) + } +} + +// getRunningJobs asks the peer for its running jobs and waits up to timeout for +// the reply (the capture side of a reload). +func (s *devSession) getRunningJobs(timeout time.Duration) (*agent.GetRunningAgentJobsResponse, error) { + ch := s.register(kindJobsResponse) + defer s.unregister(kindJobsResponse) + + err := s.write(&agent.AgentDevMessage{ + Message: &agent.AgentDevMessage_GetRunningJobsRequest{GetRunningJobsRequest: &agent.GetRunningAgentJobsRequest{}}, + }) + if err != nil { + return nil, err + } + + select { + case msg, ok := <-ch: + if !ok || msg == nil { + return nil, errDevSessionClosed + } + return msg.GetGetRunningJobsResponse(), nil + case <-time.After(timeout): + return nil, errDevSessionTimeout + } +} + +func (s *devSession) write(msg *agent.AgentDevMessage) error { + s.writeMu.Lock() + defer s.writeMu.Unlock() + return ipc.WriteProto(s.conn, msg) +} + +func (s *devSession) register(kind devMsgKind) chan *agent.AgentDevMessage { + ch := make(chan *agent.AgentDevMessage, 1) + s.mu.Lock() + s.pending[kind] = ch + s.mu.Unlock() + return ch +} + +func (s *devSession) unregister(kind devMsgKind) { + s.mu.Lock() + delete(s.pending, kind) + s.mu.Unlock() +} + +// deliver routes an inbound response to the waiting caller, if any. +func (s *devSession) deliver(kind devMsgKind, msg *agent.AgentDevMessage) { + s.mu.Lock() + ch := s.pending[kind] + delete(s.pending, kind) + s.mu.Unlock() + if ch != nil { + ch <- msg // buffered (cap 1): never blocks + } +} + +// failPending unblocks every outstanding caller when the connection closes. +func (s *devSession) failPending() { + s.mu.Lock() + for kind, ch := range s.pending { + close(ch) + delete(s.pending, kind) + } + s.mu.Unlock() +} + +func (s *devSession) close() error { + return s.conn.Close() +} diff --git a/cmd/lk/agent_dev_test.go b/cmd/lk/agent_dev_test.go new file mode 100644 index 000000000..4f10fda81 --- /dev/null +++ b/cmd/lk/agent_dev_test.go @@ -0,0 +1,206 @@ +// 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 ( + "net" + "testing" + "time" + + agent "github.com/livekit/protocol/livekit/agent" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/livekit/livekit-cli/v2/pkg/ipc" +) + +// writeMsg is a test helper that frames and sends a dev message. +func writeMsg(t *testing.T, conn net.Conn, m *agent.AgentDevMessage) { + t.Helper() + require.NoError(t, ipc.WriteProto(conn, m)) +} + +func jobsRequest() *agent.AgentDevMessage { + return &agent.AgentDevMessage{ + Message: &agent.AgentDevMessage_GetRunningJobsRequest{ + GetRunningJobsRequest: &agent.GetRunningAgentJobsRequest{}, + }, + } +} + +// TestDevSession_Restore_ServerInfo verifies the dev-channel handshake: a leading +// ServerInfo is surfaced via onServerInfo, and the following GetRunningJobsRequest +// is answered with the saved jobs. +func TestDevSession_Restore_ServerInfo(t *testing.T) { + rs := &devServer{ + savedJobs: &agent.GetRunningAgentJobsResponse{ + Jobs: []*agent.RunningAgentJobInfo{{WorkerId: "w1"}}, + }, + } + var gotName, gotURL string + rs.onServerInfo = func(agentName, url string) { gotName, gotURL = agentName, url } + + cliConn, agentConn := net.Pipe() + defer cliConn.Close() + defer agentConn.Close() + + s := rs.newSession(cliConn) + go s.run() + + // Agent sends ServerInfo, then the jobs request. + writeMsg(t, agentConn, &agent.AgentDevMessage{ + Message: &agent.AgentDevMessage_ServerInfo{ + ServerInfo: &agent.ServerInfo{AgentName: "inbound", Url: "wss://dztest2.livekit.cloud"}, + }, + }) + writeMsg(t, agentConn, jobsRequest()) + + // Should receive the saved jobs in reply. + agentConn.SetDeadline(time.Now().Add(2 * time.Second)) + resp := &agent.AgentDevMessage{} + require.NoError(t, ipc.ReadProto(agentConn, resp)) + + jobs := resp.GetGetRunningJobsResponse() + require.NotNil(t, jobs) + require.Len(t, jobs.Jobs, 1) + assert.Equal(t, "w1", jobs.Jobs[0].WorkerId) + assert.Equal(t, "inbound", gotName) + assert.Equal(t, "wss://dztest2.livekit.cloud", gotURL) +} + +// TestDevSession_Restore_NoServerInfo verifies backward compatibility: a client +// that sends only the jobs request (no ServerInfo) is still served, and +// onServerInfo is never invoked. +func TestDevSession_Restore_NoServerInfo(t *testing.T) { + rs := &devServer{savedJobs: &agent.GetRunningAgentJobsResponse{}} + called := false + rs.onServerInfo = func(string, string) { called = true } + + cliConn, agentConn := net.Pipe() + defer cliConn.Close() + defer agentConn.Close() + + s := rs.newSession(cliConn) + go s.run() + + writeMsg(t, agentConn, jobsRequest()) + + agentConn.SetDeadline(time.Now().Add(2 * time.Second)) + resp := &agent.AgentDevMessage{} + require.NoError(t, ipc.ReadProto(agentConn, resp)) + assert.NotNil(t, resp.GetGetRunningJobsResponse()) + assert.False(t, called, "onServerInfo must not fire without a ServerInfo message") +} + +// TestDevSession_TakeSavedJobs_SingleUse verifies the saved jobs are restored to +// the first asker only, then cleared. +func TestDevSession_TakeSavedJobs_SingleUse(t *testing.T) { + rs := &devServer{ + savedJobs: &agent.GetRunningAgentJobsResponse{ + Jobs: []*agent.RunningAgentJobInfo{{WorkerId: "w1"}}, + }, + } + + cliConn, agentConn := net.Pipe() + defer cliConn.Close() + defer agentConn.Close() + + s := rs.newSession(cliConn) + go s.run() + + agentConn.SetDeadline(time.Now().Add(2 * time.Second)) + + // First request gets the saved job. + writeMsg(t, agentConn, jobsRequest()) + resp := &agent.AgentDevMessage{} + require.NoError(t, ipc.ReadProto(agentConn, resp)) + require.Len(t, resp.GetGetRunningJobsResponse().GetJobs(), 1) + + // Second request gets an empty response — the read loop keeps serving. + writeMsg(t, agentConn, jobsRequest()) + resp = &agent.AgentDevMessage{} + require.NoError(t, ipc.ReadProto(agentConn, resp)) + require.NotNil(t, resp.GetGetRunningJobsResponse()) + assert.Empty(t, resp.GetGetRunningJobsResponse().GetJobs()) +} + +// TestDevSession_CaptureJobs verifies the outbound capture call: the session +// requests jobs from the peer and returns the peer's response. +func TestDevSession_CaptureJobs(t *testing.T) { + cliConn, agentConn := net.Pipe() + defer cliConn.Close() + defer agentConn.Close() + + s := newDevSession(cliConn) + go s.run() + + // Stand-in agent: answer the inbound jobs request with two jobs. + go func() { + req := &agent.AgentDevMessage{} + if ipc.ReadProto(agentConn, req) != nil { + return + } + _ = ipc.WriteProto(agentConn, &agent.AgentDevMessage{ + Message: &agent.AgentDevMessage_GetRunningJobsResponse{ + GetRunningJobsResponse: &agent.GetRunningAgentJobsResponse{ + Jobs: []*agent.RunningAgentJobInfo{{WorkerId: "a"}, {WorkerId: "b"}}, + }, + }, + }) + }() + + resp, err := s.getRunningJobs(2 * time.Second) + require.NoError(t, err) + require.Len(t, resp.GetJobs(), 2) +} + +// TestDevSession_CaptureJobs_Timeout verifies an unanswered capture call returns a +// timeout rather than blocking forever. +func TestDevSession_CaptureJobs_Timeout(t *testing.T) { + cliConn, agentConn := net.Pipe() + defer cliConn.Close() + defer agentConn.Close() + + // Drain the request but never reply. + go func() { + req := &agent.AgentDevMessage{} + _ = ipc.ReadProto(agentConn, req) + }() + + s := newDevSession(cliConn) + go s.run() + + _, err := s.getRunningJobs(100 * time.Millisecond) + assert.ErrorIs(t, err, errDevSessionTimeout) +} + +// TestDevSession_CaptureJobs_Closed verifies a pending capture call is unblocked +// when the connection closes. +func TestDevSession_CaptureJobs_Closed(t *testing.T) { + cliConn, agentConn := net.Pipe() + defer agentConn.Close() + + s := newDevSession(cliConn) + go s.run() + + go func() { + req := &agent.AgentDevMessage{} + _ = ipc.ReadProto(agentConn, req) + cliConn.Close() // drop the connection instead of replying + }() + + _, err := s.getRunningJobs(2 * time.Second) + assert.ErrorIs(t, err, errDevSessionClosed) +} diff --git a/cmd/lk/agent_reload.go b/cmd/lk/agent_reload.go deleted file mode 100644 index b5a18293e..000000000 --- a/cmd/lk/agent_reload.go +++ /dev/null @@ -1,99 +0,0 @@ -package main - -import ( - "fmt" - "net" - "sync" - "time" - - agent "github.com/livekit/protocol/livekit/agent" - - "github.com/livekit/livekit-cli/v2/pkg/ipc" -) - -// reloadServer manages the dev-mode reload protocol between Go and Python processes. -// Flow: -// 1. Go → old Python: GetRunningJobsRequest → receives GetRunningJobsResponse (capture) -// 2. New Python → Go: GetRunningJobsRequest → Go replies with saved GetRunningJobsResponse (restore) -type reloadServer struct { - listener *ipc.Listener - mu sync.Mutex - savedJobs *agent.GetRunningAgentJobsResponse -} - -func newReloadServer() (*reloadServer, error) { - ln, err := ipc.Listen("127.0.0.1:0") - if err != nil { - return nil, fmt.Errorf("reload server: %w", err) - } - return &reloadServer{listener: ln}, nil -} - -func (rs *reloadServer) addr() string { - return rs.listener.Addr().String() -} - -// captureJobs sends GetRunningJobsRequest to the old Python process and stores the response. -func (rs *reloadServer) captureJobs(conn net.Conn) { - conn.SetDeadline(time.Now().Add(1500 * time.Millisecond)) - defer conn.SetDeadline(time.Time{}) - - req := &agent.AgentDevMessage{ - Message: &agent.AgentDevMessage_GetRunningJobsRequest{ - GetRunningJobsRequest: &agent.GetRunningAgentJobsRequest{}, - }, - } - if err := ipc.WriteProto(conn, req); err != nil { - out.Warnf("reload: failed to send capture request: %v", err) - return - } - - resp := &agent.AgentDevMessage{} - if err := ipc.ReadProto(conn, resp); err != nil { - out.Warnf("reload: failed to read capture response: %v", err) - return - } - - if jobs := resp.GetGetRunningJobsResponse(); jobs != nil { - rs.mu.Lock() - rs.savedJobs = jobs - rs.mu.Unlock() - out.Statusf("reload: captured %d running job(s)", len(jobs.Jobs)) - } -} - -// serveNewProcess handles a GetRunningJobsRequest from the new Python process, -// replying with the previously captured jobs. -func (rs *reloadServer) serveNewProcess(conn net.Conn) { - req := &agent.AgentDevMessage{} - if err := ipc.ReadProto(conn, req); err != nil { - return - } - if req.GetGetRunningJobsRequest() == nil { - return - } - - rs.mu.Lock() - saved := rs.savedJobs - rs.savedJobs = nil - rs.mu.Unlock() - - if saved == nil { - saved = &agent.GetRunningAgentJobsResponse{} - } - - resp := &agent.AgentDevMessage{ - Message: &agent.AgentDevMessage_GetRunningJobsResponse{ - GetRunningJobsResponse: saved, - }, - } - if err := ipc.WriteProto(conn, resp); err != nil { - out.Warnf("reload: failed to send restore response: %v", err) - } else if len(saved.Jobs) > 0 { - out.Statusf("reload: restored %d job(s) to new process", len(saved.Jobs)) - } -} - -func (rs *reloadServer) close() error { - return rs.listener.Close() -} diff --git a/cmd/lk/agent_run.go b/cmd/lk/agent_run.go index c840eecc6..2990ddbe7 100644 --- a/cmd/lk/agent_run.go +++ b/cmd/lk/agent_run.go @@ -16,16 +16,62 @@ package main import ( "context" + "fmt" + "net/url" "os" "os/signal" + "strings" "sync" "syscall" + "time" "github.com/urfave/cli/v3" "github.com/livekit/livekit-cli/v2/pkg/agentfs" + "github.com/livekit/livekit-cli/v2/pkg/util" ) +// cloudConsoleURL returns the LiveKit Cloud agents-console URL for a worker that +// registered against wsURL with the given agent name, or "" when wsURL is not a +// LiveKit Cloud project (e.g. a self-hosted or localhost server), in which case +// no console link is shown. +func cloudConsoleURL(wsURL, agentName string) string { + consoleHost, sub := cloudProject(wsURL) + if consoleHost == "" { + return "" + } + return fmt.Sprintf( + "https://%s/projects/d_%s/agents/console?agentName=%s&autoStart=false", + consoleHost, sub, url.QueryEscape(agentName), + ) +} + +// cloudProject maps a LiveKit Cloud project URL to its console host and project +// subdomain, or ("", "") if the host is not a recognized cloud project (self- +// hosted, localhost, or any other domain). Both production and staging are +// supported: +// +// wss://my-proj.livekit.cloud -> cloud.livekit.io, my-proj +// wss://my-proj.staging.livekit.cloud -> cloud.staging.livekit.io, my-proj +func cloudProject(wsURL string) (consoleHost, subdomain string) { + u, err := url.Parse(wsURL) + if err != nil { + return "", "" + } + sub := util.ExtractSubdomain(wsURL) + if sub == "" { + return "", "" + } + host := u.Hostname() + switch { + case strings.EqualFold(host, sub+".staging.livekit.cloud"): + return "cloud.staging.livekit.io", sub + case strings.EqualFold(host, sub+".livekit.cloud"): + return "cloud.livekit.io", sub + } + return "", "" +} + func init() { AgentCommands[0].Commands = append(AgentCommands[0].Commands, startCommand, devCommand) } @@ -56,48 +102,99 @@ var devCommand = &cli.Command{ Action: runAgentDev, } +// agentCredentials holds the connection details forwarded to the agent subprocess. +type agentCredentials struct { + url string + apiKey string + apiSecret string +} + +// complete reports whether every field is populated, meaning no project lookup +// is needed to fill in the gaps. +func (c agentCredentials) complete() bool { + return c.url != "" && c.apiKey != "" && c.apiSecret != "" +} + +// args renders the credentials as agent subprocess CLI flags, skipping any empty +// field so the agent can fall back to its own defaults / environment. +func (c agentCredentials) args() []string { + var args []string + if c.url != "" { + args = append(args, "--url", c.url) + } + if c.apiKey != "" { + args = append(args, "--api-key", c.apiKey) + } + if c.apiSecret != "" { + args = append(args, "--api-secret", c.apiSecret) + } + return args +} + +// explicitCredentials returns only the credentials the user explicitly provided, +// via command-line flags or the LIVEKIT_* environment variables. +// +// The global --url flag carries a localhost default (so commands targeting a +// local server work out of the box). That default must NOT count as an explicit +// value here: if it did, it would mask the URL of the configured/default project +// and the agent would dial localhost with the project's cloud credentials. We use +// IsSet, which is true only when the flag was set from the command line or from +// LIVEKIT_URL — not when it falls back to its static default. +func explicitCredentials(cmd *cli.Command) agentCredentials { + var c agentCredentials + if cmd.IsSet("url") { + c.url = cmd.String("url") + } + if cmd.IsSet("api-key") { + c.apiKey = cmd.String("api-key") + } + if cmd.IsSet("api-secret") { + c.apiSecret = cmd.String("api-secret") + } + return c +} + +// mergeCredentials fills empty explicit fields from the project config. Explicit +// values (an intentional override) always win; the project supplies whatever the +// user did not provide — including the URL, so the configured project overrides +// the --url localhost default. +func mergeCredentials(explicit, project agentCredentials) agentCredentials { + merged := explicit + if merged.url == "" { + merged.url = project.url + } + if merged.apiKey == "" { + merged.apiKey = project.apiKey + } + if merged.apiSecret == "" { + merged.apiSecret = project.apiSecret + } + return merged +} + // resolveCredentials returns CLI args (--url, --api-key, --api-secret) for the agent subprocess. func resolveCredentials(cmd *cli.Command, loadOpts ...loadOption) ([]string, error) { - url := cmd.String("url") - if !cmd.IsSet("url") { - // Ignore the global flag's default (http://localhost:7880) so the - // project config can supply the URL. - url = "" - } - apiKey := cmd.String("api-key") - apiSecret := cmd.String("api-secret") + explicit := explicitCredentials(cmd) + merged := explicit - // Try project config if any are missing - if url == "" || apiKey == "" || apiSecret == "" { + // Only consult the project config when the user didn't fully specify the + // connection on the command line / environment. + if !explicit.complete() { opts := append([]loadOption{ignoreURL}, loadOpts...) pc, err := loadProjectDetails(cmd, opts...) if err != nil { return nil, err } if pc != nil { - if url == "" { - url = pc.URL - } - if apiKey == "" { - apiKey = pc.APIKey - } - if apiSecret == "" { - apiSecret = pc.APISecret - } + merged = mergeCredentials(explicit, agentCredentials{ + url: pc.URL, + apiKey: pc.APIKey, + apiSecret: pc.APISecret, + }) } } - var args []string - if url != "" { - args = append(args, "--url", url) - } - if apiKey != "" { - args = append(args, "--api-key", apiKey) - } - if apiSecret != "" { - args = append(args, "--api-secret", apiSecret) - } - return args, nil + return merged.args(), nil } func buildCLIArgs(projectType agentfs.ProjectType, subcmd string, cmd *cli.Command, loadOpts ...loadOption) ([]string, error) { @@ -188,6 +285,23 @@ func runAgentDev(ctx context.Context, cmd *cli.Command) error { ForwardOutput: os.Stdout, } + // When the agent reports its ServerInfo over the dev channel, print a console + // link (LiveKit Cloud projects only) the user can open to drive/debug the + // agent in the browser. Printed once, even across hot reloads (link stays valid). + var consoleLinkOnce sync.Once + cfg.OnServerInfo = func(agentName, wsURL string) { + if link := cloudConsoleURL(wsURL, agentName); link != "" { + consoleLinkOnce.Do(func() { + // Delay briefly so the link prints after the agent's own startup + // logs rather than getting buried in them. + time.AfterFunc(time.Second, func() { + out.Status("") + out.Statusf("Agent console: %s", link) + }) + }) + } + } + out.Statusf("Detected %s agent (%s in %s)", projectType.Lang(), entrypoint, projectDir) // Take over signal handling from the global NotifyContext. diff --git a/cmd/lk/agent_run_test.go b/cmd/lk/agent_run_test.go index 54537d56b..caede6548 100644 --- a/cmd/lk/agent_run_test.go +++ b/cmd/lk/agent_run_test.go @@ -15,17 +15,22 @@ package main import ( + "context" "os" "os/exec" "path/filepath" "testing" "time" + "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" + "github.com/urfave/cli/v3" "github.com/livekit/livekit-cli/v2/pkg/agentfs" ) +const localhostURL = "http://localhost:7880" + func TestAgentProcessFailSignal(t *testing.T) { if _, err := exec.LookPath("node"); err != nil { t.Skip("node not on PATH") @@ -58,3 +63,351 @@ func TestAgentProcessFailSignal(t *testing.T) { t.Fatal("Failed() did not fire on crash marker") } } + +// clearLiveKitEnv removes the LIVEKIT_* connection env vars for the duration of +// the test so flag resolution is hermetic regardless of the developer's shell. +func clearLiveKitEnv(t *testing.T) { + t.Helper() + for _, k := range []string{"LIVEKIT_URL", "LIVEKIT_API_KEY", "LIVEKIT_API_SECRET"} { + if prev, ok := os.LookupEnv(k); ok { + key, val := k, prev + require.NoError(t, os.Unsetenv(key)) + t.Cleanup(func() { _ = os.Setenv(key, val) }) + } + } +} + +// credentialFlags returns fresh instances of the connection flags, mirroring the +// production definitions in globalFlags (verified by +// TestGlobalFlagsMatchCredentialFlags). Fresh instances are required because +// urfave/cli caches parse state on the flag pointers; reusing the shared +// package-level globalFlags across multiple app.Run calls in one test binary +// would leak state between cases. Production builds the app once, so it is +// unaffected. +func credentialFlags() []cli.Flag { + return []cli.Flag{ + &cli.StringFlag{ + Name: "url", + Sources: cli.EnvVars("LIVEKIT_URL"), + Value: localhostURL, + }, + &cli.StringFlag{ + Name: "api-key", + Sources: cli.EnvVars("LIVEKIT_API_KEY"), + }, + &cli.StringFlag{ + Name: "api-secret", + Sources: cli.EnvVars("LIVEKIT_API_SECRET"), + }, + } +} + +// runWithCredentialFlags parses argv against fresh credential flags and returns +// the agentCredentials that explicitCredentials extracts inside the action. This +// exercises the actual urfave/cli wiring (localhost default, env sources, IsSet). +func runWithCredentialFlags(t *testing.T, argv []string) agentCredentials { + t.Helper() + var got agentCredentials + app := &cli.Command{ + Name: "lk", + Flags: credentialFlags(), + Action: func(_ context.Context, cmd *cli.Command) error { + got = explicitCredentials(cmd) + return nil + }, + } + require.NoError(t, app.Run(context.Background(), argv)) + return got +} + +// TestGlobalFlagsMatchCredentialFlags guards against drift between the test +// fixtures and the real global flag definitions: the production --url flag must +// keep its localhost default and the three flags must read their LIVEKIT_* env +// vars, or the fix's assumptions break. +func TestGlobalFlagsMatchCredentialFlags(t *testing.T) { + find := func(flags []cli.Flag, name string) *cli.StringFlag { + for _, f := range flags { + if sf, ok := f.(*cli.StringFlag); ok && sf.Name == name { + return sf + } + } + t.Fatalf("flag %q not found", name) + return nil + } + + urlFlag := find(globalFlags, "url") + assert.Equal(t, localhostURL, urlFlag.Value, "production --url default changed; update localhostURL") + assert.Contains(t, urlFlag.Sources.EnvKeys(), "LIVEKIT_URL") + assert.Contains(t, find(globalFlags, "api-key").Sources.EnvKeys(), "LIVEKIT_API_KEY") + assert.Contains(t, find(globalFlags, "api-secret").Sources.EnvKeys(), "LIVEKIT_API_SECRET") +} + +// --- explicitCredentials: defaults vs. intentional overrides ----------------- + +func TestExplicitCredentials_LocalhostDefaultIsNotExplicit(t *testing.T) { + // The crux of the bug: with nothing provided, the --url localhost default + // must NOT be reported as an explicit value, otherwise it masks the + // configured project's URL downstream. + clearLiveKitEnv(t) + + got := runWithCredentialFlags(t, []string{"lk"}) + + assert.Empty(t, got.url, "localhost default must not count as explicitly set") + assert.Empty(t, got.apiKey) + assert.Empty(t, got.apiSecret) +} + +func TestExplicitCredentials_URLFlagIsExplicit(t *testing.T) { + clearLiveKitEnv(t) + + got := runWithCredentialFlags(t, []string{"lk", "--url", "wss://example.livekit.cloud"}) + + assert.Equal(t, "wss://example.livekit.cloud", got.url) + assert.Empty(t, got.apiKey) + assert.Empty(t, got.apiSecret) +} + +func TestExplicitCredentials_ExplicitLocalhostFlagIsHonored(t *testing.T) { + // A user intentionally passing the localhost URL must be treated as explicit, + // even though it equals the default value. + clearLiveKitEnv(t) + + got := runWithCredentialFlags(t, []string{"lk", "--url", localhostURL}) + + assert.Equal(t, localhostURL, got.url) +} + +func TestExplicitCredentials_EnvVarsAreExplicit(t *testing.T) { + clearLiveKitEnv(t) + t.Setenv("LIVEKIT_URL", "wss://from-env.livekit.cloud") + t.Setenv("LIVEKIT_API_KEY", "envkey") + t.Setenv("LIVEKIT_API_SECRET", "envsecret") + + got := runWithCredentialFlags(t, []string{"lk"}) + + assert.Equal(t, "wss://from-env.livekit.cloud", got.url) + assert.Equal(t, "envkey", got.apiKey) + assert.Equal(t, "envsecret", got.apiSecret) +} + +func TestExplicitCredentials_AllFlags(t *testing.T) { + clearLiveKitEnv(t) + + got := runWithCredentialFlags(t, []string{ + "lk", + "--url", "wss://example.livekit.cloud", + "--api-key", "flagkey", + "--api-secret", "flagsecret", + }) + + assert.Equal(t, agentCredentials{ + url: "wss://example.livekit.cloud", + apiKey: "flagkey", + apiSecret: "flagsecret", + }, got) +} + +func TestExplicitCredentials_KeyAndSecretWithoutURL(t *testing.T) { + // Common local-server usage: key/secret supplied, URL left to the default. + // URL must remain unset so the project (or the localhost default applied by + // loadProjectDetails) can fill it in. + clearLiveKitEnv(t) + + got := runWithCredentialFlags(t, []string{ + "lk", + "--api-key", "devkey", + "--api-secret", "secret", + }) + + assert.Empty(t, got.url) + assert.Equal(t, "devkey", got.apiKey) + assert.Equal(t, "secret", got.apiSecret) +} + +// --- mergeCredentials: explicit overrides project, project fills the gaps ----- + +func TestMergeCredentials(t *testing.T) { + cloud := agentCredentials{ + url: "wss://project.livekit.cloud", + apiKey: "projkey", + apiSecret: "projsecret", + } + + tests := []struct { + name string + explicit agentCredentials + project agentCredentials + want agentCredentials + }{ + { + name: "nothing explicit -> project wins (the bug fix)", + explicit: agentCredentials{}, + project: cloud, + want: cloud, + }, + { + name: "explicit url overrides project url, project fills creds", + explicit: agentCredentials{url: localhostURL}, + project: cloud, + want: agentCredentials{url: localhostURL, apiKey: "projkey", apiSecret: "projsecret"}, + }, + { + name: "fully explicit -> project ignored", + explicit: agentCredentials{url: localhostURL, apiKey: "k", apiSecret: "s"}, + project: cloud, + want: agentCredentials{url: localhostURL, apiKey: "k", apiSecret: "s"}, + }, + { + name: "explicit creds, project supplies url", + explicit: agentCredentials{apiKey: "k", apiSecret: "s"}, + project: cloud, + want: agentCredentials{url: "wss://project.livekit.cloud", apiKey: "k", apiSecret: "s"}, + }, + { + name: "project supplies localhost url for local dev", + explicit: agentCredentials{apiKey: "devkey", apiSecret: "secret"}, + project: agentCredentials{url: localhostURL}, + want: agentCredentials{url: localhostURL, apiKey: "devkey", apiSecret: "secret"}, + }, + { + name: "empty project leaves explicit untouched", + explicit: agentCredentials{url: "wss://a", apiKey: "k", apiSecret: "s"}, + project: agentCredentials{}, + want: agentCredentials{url: "wss://a", apiKey: "k", apiSecret: "s"}, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + assert.Equal(t, tc.want, mergeCredentials(tc.explicit, tc.project)) + }) + } +} + +// --- agentCredentials.complete / args ---------------------------------------- + +func TestAgentCredentialsComplete(t *testing.T) { + assert.True(t, agentCredentials{url: "u", apiKey: "k", apiSecret: "s"}.complete()) + assert.False(t, agentCredentials{apiKey: "k", apiSecret: "s"}.complete(), "missing url") + assert.False(t, agentCredentials{url: "u", apiSecret: "s"}.complete(), "missing key") + assert.False(t, agentCredentials{url: "u", apiKey: "k"}.complete(), "missing secret") + assert.False(t, agentCredentials{}.complete()) +} + +func TestAgentCredentialsArgs(t *testing.T) { + tests := []struct { + name string + creds agentCredentials + want []string + }{ + { + name: "all set", + creds: agentCredentials{url: "u", apiKey: "k", apiSecret: "s"}, + want: []string{"--url", "u", "--api-key", "k", "--api-secret", "s"}, + }, + { + name: "empty fields are skipped", + creds: agentCredentials{apiKey: "k"}, + want: []string{"--api-key", "k"}, + }, + { + name: "nothing set -> no args", + creds: agentCredentials{}, + want: nil, + }, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + assert.Equal(t, tc.want, tc.creds.args()) + }) + } +} + +// --- resolveCredentials: end-to-end for the no-project-lookup path ------------ +// +// When the command line fully specifies the connection, resolveCredentials must +// not consult the project config at all (so it is safe to run here without a +// configured CLI config / network). The project-backed paths are covered by the +// mergeCredentials table above. + +func TestResolveCredentials_FullyExplicitSkipsProjectLookup(t *testing.T) { + clearLiveKitEnv(t) + + var args []string + app := &cli.Command{ + Name: "lk", + Flags: globalFlags, + Action: func(_ context.Context, cmd *cli.Command) error { + var err error + args, err = resolveCredentials(cmd) + return err + }, + } + + err := app.Run(context.Background(), []string{ + "lk", + "--url", "wss://example.livekit.cloud", + "--api-key", "flagkey", + "--api-secret", "flagsecret", + }) + require.NoError(t, err) + assert.Equal(t, []string{ + "--url", "wss://example.livekit.cloud", + "--api-key", "flagkey", + "--api-secret", "flagsecret", + }, args) +} + +// --- cloud console link -------------------------------------------------------- + +func TestCloudProject(t *testing.T) { + tests := []struct { + url string + wantHost string + wantSub string + }{ + {"wss://dztest2.livekit.cloud", "cloud.livekit.io", "dztest2"}, + {"https://my-proj.livekit.cloud", "cloud.livekit.io", "my-proj"}, + {"wss://dztest2.livekit.cloud:443/rtc", "cloud.livekit.io", "dztest2"}, // port + path ignored + {"wss://DZTEST2.LiveKit.Cloud", "cloud.livekit.io", "DZTEST2"}, // host match is case-insensitive + {"wss://dztest2.staging.livekit.cloud", "cloud.staging.livekit.io", "dztest2"}, + {"wss://my-proj.staging.livekit.cloud:443/rtc", "cloud.staging.livekit.io", "my-proj"}, + {"http://localhost:7880", "", ""}, // self-hosted + {"ws://192.168.1.10:7880", "", ""}, // self-hosted IP + {"wss://x.dev.livekit.cloud", "", ""}, // unrecognized multi-label host + {"wss://example.com", "", ""}, // not livekit.cloud + {"wss://livekit.cloud", "", ""}, // no subdomain + {"", "", ""}, + } + for _, tc := range tests { + t.Run(tc.url, func(t *testing.T) { + host, sub := cloudProject(tc.url) + assert.Equal(t, tc.wantHost, host) + assert.Equal(t, tc.wantSub, sub) + }) + } +} + +func TestCloudConsoleURL(t *testing.T) { + assert.Equal(t, + "https://cloud.livekit.io/projects/d_dztest2/agents/console?agentName=my-agent&autoStart=false", + cloudConsoleURL("wss://dztest2.livekit.cloud", "my-agent"), + ) + // staging projects point at the staging console host + assert.Equal(t, + "https://cloud.staging.livekit.io/projects/d_dztest2/agents/console?agentName=my-agent&autoStart=false", + cloudConsoleURL("wss://dztest2.staging.livekit.cloud", "my-agent"), + ) + // empty agent name (the common dev default) still yields a usable link + assert.Equal(t, + "https://cloud.livekit.io/projects/d_dztest2/agents/console?agentName=&autoStart=false", + cloudConsoleURL("wss://dztest2.livekit.cloud", ""), + ) + // agent names are query-escaped + assert.Equal(t, + "https://cloud.livekit.io/projects/d_dztest2/agents/console?agentName=my+agent%2F1&autoStart=false", + cloudConsoleURL("wss://dztest2.livekit.cloud", "my agent/1"), + ) + // non-cloud URLs produce no link + assert.Empty(t, cloudConsoleURL("http://localhost:7880", "my-agent")) +} diff --git a/cmd/lk/agent_watcher.go b/cmd/lk/agent_watcher.go index cd53ac3f3..03094e80e 100644 --- a/cmd/lk/agent_watcher.go +++ b/cmd/lk/agent_watcher.go @@ -16,7 +16,6 @@ package main import ( "fmt" - "net" "os" "path/filepath" "strings" @@ -51,8 +50,8 @@ type agentWatcher struct { agent *AgentProcess restartCh chan struct{} - reloadSrv *reloadServer - conn net.Conn + devSrv *devServer + session *devSession } func newAgentWatcher(config AgentStartConfig) (*agentWatcher, error) { @@ -82,9 +81,9 @@ func newAgentWatcher(config AgentStartConfig) (*agentWatcher, error) { // The reload protocol (capture running jobs from the old process, restore // them in the new one) is Python-only; Node reloads are a plain kill+respawn. - var rs *reloadServer + var rs *devServer if config.ProjectType.IsPython() { - rs, err = newReloadServer() + rs, err = newDevServer() if err != nil { w.Close() return nil, err @@ -92,6 +91,7 @@ func newAgentWatcher(config AgentStartConfig) (*agentWatcher, error) { // Append --reload-addr to CLI args so the Python process connects back config.CLIArgs = append(config.CLIArgs, "--reload-addr", rs.addr()) } + rs.onServerInfo = config.OnServerInfo return &agentWatcher{ config: config, @@ -99,7 +99,7 @@ func newAgentWatcher(config AgentStartConfig) (*agentWatcher, error) { debounce: 500 * time.Millisecond, watcher: w, restartCh: make(chan struct{}, 1), - reloadSrv: rs, + devSrv: rs, }, nil } @@ -110,28 +110,34 @@ func (aw *agentWatcher) start() error { } aw.agent = agent - // Accept connection from new Python process in background - if aw.reloadSrv != nil { - go func() { - conn, err := aw.reloadSrv.listener.Accept() - if err != nil { - return - } - aw.conn = conn - // Serve the initial restore request (will be empty on first start) - go aw.reloadSrv.serveNewProcess(conn) - }() - } + aw.acceptSession() return nil } +// acceptSession waits (in the background) for the next process to connect back on +// the reload channel and hands the connection to a devSession read loop. +func (aw *agentWatcher) acceptSession() { + if aw.devSrv == nil { + return + } + go func() { + conn, err := aw.devSrv.listener.Accept() + if err != nil { + return + } + s := aw.devSrv.newSession(conn) + aw.session = s + go s.run() + }() +} + func (aw *agentWatcher) restart() error { // 1. Capture active jobs from the current process (best-effort) - if aw.conn != nil { - aw.reloadSrv.captureJobs(aw.conn) - aw.conn.Close() - aw.conn = nil + if aw.session != nil { + aw.devSrv.captureJobs(aw.session) + aw.session.close() + aw.session = nil } // 2. Kill old process @@ -149,16 +155,7 @@ func (aw *agentWatcher) restart() error { aw.agent = agent // 4. Accept new connection and serve restored jobs - if aw.reloadSrv != nil { - go func() { - conn, err := aw.reloadSrv.listener.Accept() - if err != nil { - return - } - aw.conn = conn - go aw.reloadSrv.serveNewProcess(conn) - }() - } + aw.acceptSession() return nil } @@ -181,11 +178,11 @@ func (aw *agentWatcher) Run(done <-chan struct{}) error { aw.agent.ForceKill() } } - if aw.conn != nil { - aw.conn.Close() + if aw.session != nil { + aw.session.close() } - if aw.reloadSrv != nil { - aw.reloadSrv.close() + if aw.devSrv != nil { + aw.devSrv.close() } aw.watcher.Close() }() diff --git a/cmd/lk/simulate_subprocess.go b/cmd/lk/simulate_subprocess.go index 979949fa6..dbbf50187 100644 --- a/cmd/lk/simulate_subprocess.go +++ b/cmd/lk/simulate_subprocess.go @@ -268,6 +268,11 @@ type AgentStartConfig struct { ReadySignal string // substring to scan for in output (e.g. "registered worker"), empty to skip FailSignals []string // output substrings meaning the agent has fatally failed even if the process is still alive ForwardOutput io.Writer // if set, forward each output line to this writer + + // OnServerInfo, if set, is invoked when the agent reports its ServerInfo over + // the dev-reload channel (agent name + the LiveKit URL it uses). Only wired + // for the reload-based dev path, which owns the channel. + OnServerInfo func(agentName, wsURL string) } // thinCLIMinVersion is the first livekit-agents release that exposes the diff --git a/go.mod b/go.mod index d9feed033..af90c7e09 100644 --- a/go.mod +++ b/go.mod @@ -19,15 +19,15 @@ require ( github.com/google/go-containerregistry v0.20.7 github.com/google/go-querystring v1.2.0 github.com/joho/godotenv v1.5.1 - github.com/livekit/protocol v1.47.0 - github.com/livekit/server-sdk-go/v2 v2.16.7-0.20260608025623-a5da15b13baa + github.com/livekit/protocol v1.48.1-0.20260624204523-bd5703442db6 + github.com/livekit/server-sdk-go/v2 v2.16.7 github.com/mattn/go-isatty v0.0.22 github.com/moby/patternmatcher v0.6.1 github.com/modelcontextprotocol/go-sdk v1.6.1 github.com/pelletier/go-toml v1.9.5 github.com/pion/rtcp v1.2.16 github.com/pion/rtp v1.10.2 - github.com/pion/webrtc/v4 v4.2.11 + github.com/pion/webrtc/v4 v4.2.14 github.com/pkg/browser v0.0.0-20240102092130-5ac0b6a4141c github.com/stretchr/testify v1.11.1 github.com/twitchtv/twirp v8.1.3+incompatible @@ -101,7 +101,7 @@ require ( github.com/cncf/xds/go v0.0.0-20260202195803-dba9d589def2 // indirect github.com/containerd/console v1.0.5 // indirect github.com/containerd/containerd/api v1.10.0 // indirect - github.com/containerd/containerd/v2 v2.2.3 // indirect + github.com/containerd/containerd/v2 v2.2.4 // indirect github.com/containerd/continuity v0.4.5 // indirect github.com/containerd/errdefs v1.0.0 // indirect github.com/containerd/errdefs/pkg v0.3.0 // indirect @@ -186,13 +186,13 @@ require ( github.com/opencontainers/image-spec v1.1.1 // indirect github.com/pierrec/lz4/v4 v4.1.26 // indirect github.com/pion/datachannel v1.6.0 // indirect - github.com/pion/dtls/v3 v3.1.3 // indirect + github.com/pion/dtls/v3 v3.1.4 // indirect github.com/pion/ice/v4 v4.2.7 // indirect github.com/pion/interceptor v0.1.45 // indirect github.com/pion/logging v0.2.4 // indirect github.com/pion/mdns/v2 v2.1.0 // indirect github.com/pion/randutil v0.1.0 // indirect - github.com/pion/sctp v1.9.5 // indirect + github.com/pion/sctp v1.10.0 // indirect github.com/pion/sdp/v3 v3.0.18 // indirect github.com/pion/srtp/v3 v3.0.11 // indirect github.com/pion/stun/v3 v3.1.4 // indirect @@ -264,3 +264,7 @@ require ( mvdan.cc/sh/v3 v3.13.2-0.20260510185049-f5c6e2779117 // indirect sigs.k8s.io/json v0.0.0-20250730193827-2d320260d730 // indirect ) + +// TEMP (local dev): use local protocol with the WorkerInfo dev message. +// Drop once github.com/livekit/protocol publishes it. +// replace github.com/livekit/protocol => ../protocol diff --git a/go.sum b/go.sum index 7c6c080d0..fb153433c 100644 --- a/go.sum +++ b/go.sum @@ -184,8 +184,8 @@ github.com/containerd/console v1.0.5 h1:R0ymNeydRqH2DmakFNdmjR2k0t7UPuiOV/N/27/q github.com/containerd/console v1.0.5/go.mod h1:YynlIjWYF8myEu6sdkwKIvGQq+cOckRm6So2avqoYAk= github.com/containerd/containerd/api v1.10.0 h1:5n0oHYVBwN4VhoX9fFykCV9dF1/BvAXeg2F8W6UYq1o= github.com/containerd/containerd/api v1.10.0/go.mod h1:NBm1OAk8ZL+LG8R0ceObGxT5hbUYj7CzTmR3xh0DlMM= -github.com/containerd/containerd/v2 v2.2.3 h1:mOBRLaHGvmgy0bRo1Sg6OD8ugMKZIvCoWWMeMMygliA= -github.com/containerd/containerd/v2 v2.2.3/go.mod h1:ns24cwt+p36mRnuKE3hLRxVBpuSP+a/Y25AMki1t/RY= +github.com/containerd/containerd/v2 v2.2.4 h1:8x2UdXqww7NYqGNabQ7i1nAgB5LegzjC9KQzO/900iA= +github.com/containerd/containerd/v2 v2.2.4/go.mod h1:YBcTO8D9149QY9zNmUjy04Mhuc4DlrZQ8FIOwKZEM7o= github.com/containerd/continuity v0.4.5 h1:ZRoN1sXq9u7V6QoHMcVWGhOwDFqZ4B9i5H6un1Wh0x4= github.com/containerd/continuity v0.4.5/go.mod h1:/lNJvtJKUQStBzpVQ1+rasXO1LAWtUQssk28EZvJ3nE= github.com/containerd/errdefs v1.0.0 h1:tg5yIfIlQIrxYtu9ajqY42W3lpS19XqdxRQeEwYG8PI= @@ -357,12 +357,12 @@ github.com/livekit/mageutil v0.0.0-20250511045019-0f1ff63f7731 h1:9x+U2HGLrSw5AT github.com/livekit/mageutil v0.0.0-20250511045019-0f1ff63f7731/go.mod h1:Rs3MhFwutWhGwmY1VQsygw28z5bWcnEYmS1OG9OxjOQ= github.com/livekit/mediatransportutil v0.0.0-20260605212259-862d4a7bcb1e h1:SkgQRcG2VYEhh80Qb/zYZo8rWKJzNfJcfUQnXe6su2M= github.com/livekit/mediatransportutil v0.0.0-20260605212259-862d4a7bcb1e/go.mod h1:o8CFmAdrVwzJNOCsQCLUzXRjokkufNshnQHOe4fRaqU= -github.com/livekit/protocol v1.47.0 h1:6dwpf2pSRnvUlhpYyVESFQiBCj8klFKbC40bZjX7AwY= -github.com/livekit/protocol v1.47.0/go.mod h1:jO+y05AU9Ec4JswDyuzKCZ4bhziOS0CzMqgnbj60Dzs= +github.com/livekit/protocol v1.48.1-0.20260624204523-bd5703442db6 h1:Z23fxGEJJS3akyX2Q6LGo5xfL+B+NcEBhOp5o0Cz5IE= +github.com/livekit/protocol v1.48.1-0.20260624204523-bd5703442db6/go.mod h1:jO+y05AU9Ec4JswDyuzKCZ4bhziOS0CzMqgnbj60Dzs= github.com/livekit/psrpc v0.7.2 h1:6oZ+NODJ2pLyaT6VqDq1F4Qc/3TpDUSpyphj/P9MhQc= github.com/livekit/psrpc v0.7.2/go.mod h1:rAI+m2+/cb4x9RXhLRtUx5ZwdfjjXOl4zi46IjEetaw= -github.com/livekit/server-sdk-go/v2 v2.16.7-0.20260608025623-a5da15b13baa h1:B19yilP7+JjekKMD0WejMh1Kvypdxpr5yxQZiFStRD0= -github.com/livekit/server-sdk-go/v2 v2.16.7-0.20260608025623-a5da15b13baa/go.mod h1:SWJD68Rfcwrhze09EYaRiur7ESCBuu0u4fpK+0BGEYo= +github.com/livekit/server-sdk-go/v2 v2.16.7 h1:oYnp2o3YTBdL4xVkq5NpLwqnCijZVfVJU9ddFl+BW/Y= +github.com/livekit/server-sdk-go/v2 v2.16.7/go.mod h1:B3qlhVBZ4olBWRN/KxokLZE2d4LujNk4n2yYDL3+u2s= github.com/lucasb-eyer/go-colorful v1.4.0 h1:UtrWVfLdarDgc44HcS7pYloGHJUjHV/4FwW4TvVgFr4= github.com/lucasb-eyer/go-colorful v1.4.0/go.mod h1:R4dSotOR9KMtayYi1e77YzuveK+i7ruzyGqttikkLy0= github.com/magefile/mage v1.17.2 h1:fyXVu1eadI8Ap1HCCNgEhJ5McIWiYhLR8uol64ZZc40= @@ -445,8 +445,8 @@ github.com/pierrec/lz4/v4 v4.1.26 h1:GrpZw1gZttORinvzBdXPUXATeqlJjqUG/D87TKMnhjY github.com/pierrec/lz4/v4 v4.1.26/go.mod h1:EoQMVJgeeEOMsCqCzqFm2O0cJvljX2nGZjcRIPL34O4= github.com/pion/datachannel v1.6.0 h1:XecBlj+cvsxhAMZWFfFcPyUaDZtd7IJvrXqlXD/53i0= github.com/pion/datachannel v1.6.0/go.mod h1:ur+wzYF8mWdC+Mkis5Thosk+u/VOL287apDNEbFpsIk= -github.com/pion/dtls/v3 v3.1.3 h1:OA6J5UCeA8DvRXD8ofaMnlNPXN3ISBLHHJ9P8SWL09E= -github.com/pion/dtls/v3 v3.1.3/go.mod h1:GEwid4EzCcakfrNvHXM7bs6ci2mASI5Y5Q4tbtLFuWs= +github.com/pion/dtls/v3 v3.1.4 h1:QhvtMflMfu9Kf0RcDC5BJBle4caPskByrKQR6uuYqpY= +github.com/pion/dtls/v3 v3.1.4/go.mod h1:cr/qotLISUw/9C1m83ZPNZtj9WnXkYLpfCptPqbkInc= github.com/pion/ice/v4 v4.2.7 h1:zDEbC6MiEdhQpF8TxBOTws+NU6ZgGpveHrQq4Lc1kao= github.com/pion/ice/v4 v4.2.7/go.mod h1:9SNPaq0c7El/ki8leJzyCkK10zsskprR3zTNbO3monY= github.com/pion/interceptor v0.1.45 h1:6PUo/5829bIfRFIPPJQzuDn8EjxRTSB/CSD7QVCOaqo= @@ -461,8 +461,8 @@ github.com/pion/rtcp v1.2.16 h1:fk1B1dNW4hsI78XUCljZJlC4kZOPk67mNRuQ0fcEkSo= github.com/pion/rtcp v1.2.16/go.mod h1:/as7VKfYbs5NIb4h6muQ35kQF/J0ZVNz2Z3xKoCBYOo= github.com/pion/rtp v1.10.2 h1:l+f6tTDcAH6xwepaAoW791ddhuYsJlqRATOzirO04Mo= github.com/pion/rtp v1.10.2/go.mod h1:Au8fc6cEByy8RLTwKTQTEeQqDB/SJDxwL4mZuxYA5Pk= -github.com/pion/sctp v1.9.5 h1:QoSFB/drmAsmSeSFNQNI3xx010nW4HsycCZckRVWWag= -github.com/pion/sctp v1.9.5/go.mod h1:N20Dq6LY+JvJDAh9VVh1JELngb2rQ8dPgds5yBWiPgw= +github.com/pion/sctp v1.10.0 h1:qeoD6swF/2M5bYRcAGayqSbTKX3m4AW29CiQxG1+Pfg= +github.com/pion/sctp v1.10.0/go.mod h1:N20Dq6LY+JvJDAh9VVh1JELngb2rQ8dPgds5yBWiPgw= github.com/pion/sdp/v3 v3.0.18 h1:l0bAXazKHpepazVdp+tPYnrsy9dfh7ZbT8DxesH5ZnI= github.com/pion/sdp/v3 v3.0.18/go.mod h1:ZREGo6A9ZygQ9XkqAj5xYCQtQpif0i6Pa81HOiAdqQ8= github.com/pion/srtp/v3 v3.0.11 h1:GiESUr54/K4UuPigfq/CvWUed80JenQAHXn0C2MQQIQ= @@ -473,12 +473,10 @@ github.com/pion/transport/v3 v3.1.1 h1:Tr684+fnnKlhPceU+ICdrw6KKkTms+5qHMgw6bIkY github.com/pion/transport/v3 v3.1.1/go.mod h1:+c2eewC5WJQHiAA46fkMMzoYZSuGzA/7E2FPrOYHctQ= github.com/pion/transport/v4 v4.0.2 h1:ifYlPqNwsy6aKQ9y8yzxXlHae5431ZrH2avkD/Rn6Tk= github.com/pion/transport/v4 v4.0.2/go.mod h1:06hFI+jCFcok2X2MekVufNZ/uzNZXivGBPfviSVcjgM= -github.com/pion/turn/v4 v4.1.4 h1:EU11yMXKIsK43FhcUnjLlrhE4nboHZq+TXBIi3QpcxQ= -github.com/pion/turn/v4 v4.1.4/go.mod h1:ES1DXVFKnOhuDkqn9hn5VJlSWmZPaRJLyBXoOeO/BmQ= github.com/pion/turn/v5 v5.0.8 h1:pZUCtmwWCMkrRKqh/8pL3WoGADXBe0/lOPkN7oqFjK8= github.com/pion/turn/v5 v5.0.8/go.mod h1:1VwvxElZaOdJU0liJ/WUSm/Tsh+n2OxS5ISSDxgOWxU= -github.com/pion/webrtc/v4 v4.2.11 h1:QUX1QZKlNIn4O7U5JxLPGP0sV5RTncZkzu9SPR3jVNU= -github.com/pion/webrtc/v4 v4.2.11/go.mod h1:s/rAiyy77GyRFrZMx+Ls6aua26dIBPudH8/ZHYbIRWY= +github.com/pion/webrtc/v4 v4.2.14 h1:Q6zMs+fSDsYuhZcNlvFGBxCOMHVV9oYcDa6O9/HIGTc= +github.com/pion/webrtc/v4 v4.2.14/go.mod h1:87NVKP86+g4OMrRxWhjWfUjeXP4JrV6RTlUrIW+/Jak= github.com/pkg/browser v0.0.0-20240102092130-5ac0b6a4141c h1:+mdjkGKdHQG3305AYmdv1U2eRNDiU2ErMBj1gwrq8eQ= github.com/pkg/browser v0.0.0-20240102092130-5ac0b6a4141c/go.mod h1:7rwL4CYBLnjLxUqIJNnCWiEdr3bn6IUYi15bNlnbCCU= github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4= From 3c617de2914d11737c145186f028952084c934a9 Mon Sep 17 00:00:00 2001 From: David Zhao Date: Wed, 8 Jul 2026 23:32:19 +0200 Subject: [PATCH 2/4] wrap url under OSC 8 --- cmd/lk/agent_run.go | 8 +++++++- pkg/util/printer.go | 7 +++++++ pkg/util/theme.go | 8 ++++++++ 3 files changed, 22 insertions(+), 1 deletion(-) diff --git a/cmd/lk/agent_run.go b/cmd/lk/agent_run.go index 2990ddbe7..e490ea729 100644 --- a/cmd/lk/agent_run.go +++ b/cmd/lk/agent_run.go @@ -295,8 +295,14 @@ func runAgentDev(ctx context.Context, cmd *cli.Command) error { // Delay briefly so the link prints after the agent's own startup // logs rather than getting buried in them. time.AfterFunc(time.Second, func() { + // Accent-colored, and a clickable OSC 8 hyperlink on terminals + // that support it (gated so the escape never leaks into pipes). + label := util.Accented(link) + if out.Interactive() { + label = util.Hyperlink(link, label) + } out.Status("") - out.Statusf("Agent console: %s", link) + out.Statusf("Agent console: %s", label) }) }) } diff --git a/pkg/util/printer.go b/pkg/util/printer.go index 46677290e..58cb14bfc 100644 --- a/pkg/util/printer.go +++ b/pkg/util/printer.go @@ -55,6 +55,13 @@ func NewPrinter(out, err io.Writer, quiet bool) *Printer { return &Printer{Out: out, Err: err, Quiet: quiet, interactive: isTerminal(err)} } +// Interactive reports whether status output is going to a real terminal, so +// callers can gate terminal-only escapes (e.g. OSC 8 hyperlinks) and avoid +// leaking them into piped or redirected output. +func (p *Printer) Interactive() bool { + return p != nil && p.interactive +} + // isTerminal reports whether w is a terminal-backed *os.File. Non-file writers // (bytes.Buffer in tests, pipes) are never terminals. func isTerminal(w io.Writer) bool { diff --git a/pkg/util/theme.go b/pkg/util/theme.go index 874e93033..7c42b7c6c 100644 --- a/pkg/util/theme.go +++ b/pkg/util/theme.go @@ -167,6 +167,14 @@ func Dimmed(text string) string { return Theme.Focused.Description.Render(text) } +// Hyperlink wraps label in an OSC 8 terminal hyperlink pointing at url. Terminals +// that support OSC 8 render label as a clickable link; others ignore the escape +// and show label unchanged. Gate calls on an interactive terminal (see +// Printer.Interactive) so the escape never leaks into piped/redirected output. +func Hyperlink(url, label string) string { + return "\x1b]8;;" + url + "\x1b\\" + label + "\x1b]8;;\x1b\\" +} + // Confirm is a yes/no select styled by the active theme. func Confirm() *huh.Select[bool] { return huh.NewSelect[bool](). From e02c84f449a26ee911785d2687bca4b65269d038 Mon Sep 17 00:00:00 2001 From: David Zhao Date: Wed, 8 Jul 2026 23:52:24 +0200 Subject: [PATCH 3/4] fix tab completion --- cmd/lk/agent_run.go | 6 ++++++ cmd/lk/console.go | 3 +++ cmd/lk/simulate.go | 3 +++ 3 files changed, 12 insertions(+) diff --git a/cmd/lk/agent_run.go b/cmd/lk/agent_run.go index e490ea729..95acfb5a0 100644 --- a/cmd/lk/agent_run.go +++ b/cmd/lk/agent_run.go @@ -89,6 +89,10 @@ var startCommand = &cli.Command{ ArgsUsage: "[entrypoint] [-- node/python-args...]", Flags: agentRunFlags, Action: runAgentStart, + // Leaf command whose positional arg is an entrypoint file. Drop the implicit + // `help` subcommand so shell completion returns nothing for the positional, + // letting the shell fall back to native filename completion (--help kept). + HideHelpCommand: true, } var devCommand = &cli.Command{ @@ -100,6 +104,8 @@ var devCommand = &cli.Command{ Usage: "Disable auto-reload on file changes", }), Action: runAgentDev, + // See startCommand: hide `help` so filenames complete for the entrypoint arg. + HideHelpCommand: true, } // agentCredentials holds the connection details forwarded to the agent subprocess. diff --git a/cmd/lk/console.go b/cmd/lk/console.go index affa85e2c..87cec2177 100644 --- a/cmd/lk/console.go +++ b/cmd/lk/console.go @@ -43,6 +43,9 @@ var consoleCommand = &cli.Command{ Usage: "Voice chat with an agent via mic/speakers", ArgsUsage: "[entrypoint] [-- node/python-args...]", Category: "Core", + // Hide the implicit `help` subcommand so shell completion falls back to + // native filename completion for the entrypoint arg (see startCommand). + HideHelpCommand: true, Flags: []cli.Flag{ &cli.IntFlag{ Name: "port", diff --git a/cmd/lk/simulate.go b/cmd/lk/simulate.go index 4abeb2049..a9b8fd5bd 100644 --- a/cmd/lk/simulate.go +++ b/cmd/lk/simulate.go @@ -57,6 +57,9 @@ var simulateCommand = &cli.Command{ Name: "simulate", Usage: "Run agent simulations against LiveKit Cloud", ArgsUsage: "[entrypoint]", + // Hide the implicit `help` subcommand so shell completion falls back to + // native filename completion for the entrypoint arg (see startCommand). + HideHelpCommand: true, Before: func(ctx context.Context, cmd *cli.Command) (context.Context, error) { pc, err := loadProjectDetails(cmd) if err != nil { From 4f919f8fb6987defb0b10e1786ac714c04406d0a Mon Sep 17 00:00:00 2001 From: David Zhao Date: Thu, 9 Jul 2026 00:02:08 +0200 Subject: [PATCH 4/4] fix CI --- autocomplete/fish_autocomplete | 4 ---- 1 file changed, 4 deletions(-) diff --git a/autocomplete/fish_autocomplete b/autocomplete/fish_autocomplete index abab0de0e..d3ba3a25d 100644 --- a/autocomplete/fish_autocomplete +++ b/autocomplete/fish_autocomplete @@ -168,12 +168,10 @@ complete -x -c lk -n '__fish_seen_subcommand_from agent a; and __fish_seen_subco complete -x -c lk -n '__fish_seen_subcommand_from agent a; and not __fish_seen_subcommand_from init create dockerfile config deploy status update restart rollback logs tail delete destroy versions list secrets update-secrets private-link start dev console daemon simulate help h' -a 'start' -d 'Run an agent in production mode' complete -c lk -n '__fish_seen_subcommand_from agent a; and __fish_seen_subcommand_from start' -f -l log-level -r -d 'Log level (TRACE, DEBUG, INFO, WARN, ERROR)' complete -c lk -n '__fish_seen_subcommand_from agent a; and __fish_seen_subcommand_from start' -f -l help -s h -d 'show help' -complete -x -c lk -n '__fish_seen_subcommand_from agent a; and __fish_seen_subcommand_from start; and not __fish_seen_subcommand_from help h' -a 'help' -d 'Shows a list of commands or help for one command' complete -x -c lk -n '__fish_seen_subcommand_from agent a; and not __fish_seen_subcommand_from init create dockerfile config deploy status update restart rollback logs tail delete destroy versions list secrets update-secrets private-link start dev console daemon simulate help h' -a 'dev' -d 'Run an agent in development mode with auto-reload' complete -c lk -n '__fish_seen_subcommand_from agent a; and __fish_seen_subcommand_from dev' -f -l log-level -r -d 'Log level (TRACE, DEBUG, INFO, WARN, ERROR)' complete -c lk -n '__fish_seen_subcommand_from agent a; and __fish_seen_subcommand_from dev' -f -l no-reload -d 'Disable auto-reload on file changes' complete -c lk -n '__fish_seen_subcommand_from agent a; and __fish_seen_subcommand_from dev' -f -l help -s h -d 'show help' -complete -x -c lk -n '__fish_seen_subcommand_from agent a; and __fish_seen_subcommand_from dev; and not __fish_seen_subcommand_from help h' -a 'help' -d 'Shows a list of commands or help for one command' complete -x -c lk -n '__fish_seen_subcommand_from agent a; and not __fish_seen_subcommand_from init create dockerfile config deploy status update restart rollback logs tail delete destroy versions list secrets update-secrets private-link start dev console daemon simulate help h' -a 'console' -d 'Voice chat with an agent via mic/speakers' complete -c lk -n '__fish_seen_subcommand_from agent a; and __fish_seen_subcommand_from console' -f -l port -s p -r -d 'TCP port for agent communication' complete -c lk -n '__fish_seen_subcommand_from agent a; and __fish_seen_subcommand_from console' -f -l input-device -r -d 'Input device index or name substring' @@ -183,7 +181,6 @@ complete -c lk -n '__fish_seen_subcommand_from agent a; and __fish_seen_subcomma complete -c lk -n '__fish_seen_subcommand_from agent a; and __fish_seen_subcommand_from console' -f -l text -s t -d 'Start in text mode instead of audio mode' complete -c lk -n '__fish_seen_subcommand_from agent a; and __fish_seen_subcommand_from console' -f -l record -d 'Record audio and session report to console-recordings/' complete -c lk -n '__fish_seen_subcommand_from agent a; and __fish_seen_subcommand_from console' -f -l help -s h -d 'show help' -complete -x -c lk -n '__fish_seen_subcommand_from agent a; and __fish_seen_subcommand_from console; and not __fish_seen_subcommand_from help h' -a 'help' -d 'Shows a list of commands or help for one command' complete -x -c lk -n '__fish_seen_subcommand_from agent a; and not __fish_seen_subcommand_from init create dockerfile config deploy status update restart rollback logs tail delete destroy versions list secrets update-secrets private-link start dev console daemon simulate help h' -a 'daemon' -d 'Drive a single local agent session in text mode (start/say/stop)' complete -c lk -n '__fish_seen_subcommand_from agent a; and __fish_seen_subcommand_from daemon' -f -l help -s h -d 'show help' complete -x -c lk -n '__fish_seen_subcommand_from agent a; and __fish_seen_subcommand_from daemon; and not __fish_seen_subcommand_from start say stop serve help h' -a 'start' -d 'Start a detached agent session daemon' @@ -209,7 +206,6 @@ complete -c lk -n '__fish_seen_subcommand_from agent a; and __fish_seen_subcomma complete -c lk -n '__fish_seen_subcommand_from agent a; and __fish_seen_subcommand_from simulate' -f -l yes -s y -d 'Skip the source-upload confirmation prompt (required for non-interactive runs that generate from source)' complete -c lk -n '__fish_seen_subcommand_from agent a; and __fish_seen_subcommand_from simulate' -f -l view -r -d 'Open a pre-existing simulation' complete -c lk -n '__fish_seen_subcommand_from agent a; and __fish_seen_subcommand_from simulate' -f -l help -s h -d 'show help' -complete -x -c lk -n '__fish_seen_subcommand_from agent a; and __fish_seen_subcommand_from simulate; and not __fish_seen_subcommand_from help h' -a 'help' -d 'Shows a list of commands or help for one command' complete -x -c lk -n '__fish_seen_subcommand_from agent a; and not __fish_seen_subcommand_from init create dockerfile config deploy status update restart rollback logs tail delete destroy versions list secrets update-secrets private-link start dev console daemon simulate help h' -a 'help' -d 'Shows a list of commands or help for one command' complete -x -c lk -n '__fish_lk_no_subcommand' -a 'cloud' -d 'Interact with LiveKit Cloud services' complete -c lk -n '__fish_seen_subcommand_from cloud' -f -l help -s h -d 'show help'