Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
33 changes: 19 additions & 14 deletions cmd/lk/simulate_health.go
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@ package main
import (
"regexp"
"strings"
"time"

"github.com/livekit/protocol/livekit"
)
Expand All @@ -40,27 +41,31 @@ const minFatalMarkers = 2
const maxAgentNotJoined = 3

// agentBroken reports whether the worker is failing systemically. A completed
// scenario proves the agent works.
// scenario proves the agent worked until that scenario ended, not after it: an
// agent that dies or wedges mid-run fails every remaining job on join.
func agentBroken(run *livekit.SimulationRun, ap *AgentProcess) bool {
completed, notJoined := 0, 0
var lastCompleted time.Time
completed := false
for _, job := range run.GetJobs() {
switch job.GetStatus() {
case livekit.SimulationRun_Job_STATUS_COMPLETED:
completed++
case livekit.SimulationRun_Job_STATUS_FAILED:
if strings.Contains(job.GetError(), "No agent joined room") {
notJoined++
if job.GetStatus() == livekit.SimulationRun_Job_STATUS_COMPLETED {
completed = true
if t := job.GetEndedAt().AsTime(); t.After(lastCompleted) {
lastCompleted = t
}
}
}
switch {
case completed > 0:
return false
case ap != nil && countFatalMarkers(ap.RecentLogs(0)) >= minFatalMarkers:
notJoined := 0
for _, job := range run.GetJobs() {
if job.GetStatus() == livekit.SimulationRun_Job_STATUS_FAILED &&
strings.Contains(job.GetError(), "No agent joined room") &&
job.GetEndedAt().AsTime().After(lastCompleted) {
notJoined++
}
}
if !completed && ap != nil && countFatalMarkers(ap.RecentLogs(0)) >= minFatalMarkers {
return true
default:
return notJoined > maxAgentNotJoined
}
return notJoined > maxAgentNotJoined
}

// agentErrorContext is the worker output to surface for a broken agent: from
Expand Down
42 changes: 42 additions & 0 deletions cmd/lk/simulate_health_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -16,9 +16,11 @@ package main

import (
"testing"
"time"

"github.com/livekit/protocol/livekit"
"github.com/stretchr/testify/require"
"google.golang.org/protobuf/types/known/timestamppb"
)

// Real log lines captured from a rate-limited `lk agent simulate` run.
Expand Down Expand Up @@ -91,3 +93,43 @@ func TestRunningJobCount(t *testing.T) {
require.Equal(t, 2, runningJobCount(run))
require.Equal(t, 0, runningJobCount(nil))
}

func TestAgentBroken(t *testing.T) {
t0 := time.Date(2026, 10, 15, 20, 0, 0, 0, time.UTC)
completed := func(min int) *livekit.SimulationRun_Job {
return &livekit.SimulationRun_Job{
Status: livekit.SimulationRun_Job_STATUS_COMPLETED,
EndedAt: timestamppb.New(t0.Add(time.Duration(min) * time.Minute)),
}
}
notJoined := func(min int) *livekit.SimulationRun_Job {
return &livekit.SimulationRun_Job{
Status: livekit.SimulationRun_Job_STATUS_FAILED,
Error: "No agent joined room 'sim-x' within 30s. Is the agent running?",
EndedAt: timestamppb.New(t0.Add(time.Duration(min) * time.Minute)),
}
}
run := func(jobs ...*livekit.SimulationRun_Job) *livekit.SimulationRun {
return &livekit.SimulationRun{Jobs: jobs}
}

tests := []struct {
name string
run *livekit.SimulationRun
want bool
}{
{"never joined", run(notJoined(1), notJoined(1), notJoined(1), notJoined(1)), true},
{"a few transient timeouts", run(notJoined(1), notJoined(1), notJoined(1)), false},
{"cold-start timeouts before the agent warmed up",
run(notJoined(1), notJoined(1), notJoined(1), notJoined(1), completed(2)), false},
{"agent died after working",
run(completed(1), completed(2), notJoined(3), notJoined(4), notJoined(5), notJoined(6)), true},
{"one timeout between successes",
run(completed(1), notJoined(2), completed(3), notJoined(4)), false},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
require.Equal(t, tt.want, agentBroken(tt.run, nil))
})
}
}
Loading