diff --git a/cmd/lk/simulate_health.go b/cmd/lk/simulate_health.go index 197a9f8e..4759f5ef 100644 --- a/cmd/lk/simulate_health.go +++ b/cmd/lk/simulate_health.go @@ -17,6 +17,7 @@ package main import ( "regexp" "strings" + "time" "github.com/livekit/protocol/livekit" ) @@ -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 diff --git a/cmd/lk/simulate_health_test.go b/cmd/lk/simulate_health_test.go index 08e40fa9..fa51f8a7 100644 --- a/cmd/lk/simulate_health_test.go +++ b/cmd/lk/simulate_health_test.go @@ -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. @@ -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)) + }) + } +}