fix(scheduler): key runner teardown on actual assignment, not dispatch intent

cmd/ephemerd/main.go

@@ -600,6 +600,19 @@ func serve(ctx context.Context, configFile, imagesDirFlag string, containerdTCPP
 			"rate_aware", retryCfg.RateHint != nil)
 	}
 
+	// Orphaned-runner sweep: destroys dispatched runners that were never
+	// assigned a job within the grace window. GitHub schedules JIT
+	// runners onto any queued job with matching labels, so runner
+	// teardown is keyed to observed assignments; the sweep is the
+	// cleanup path for runners whose intended job ran elsewhere.
+	orphanCfg := scheduler.OrphanSweepConfig{
+		Enabled: cfg.Runner.OrphanSweepEnabled(),
+		Grace:   cfg.Runner.OrphanSweepGrace(),
+	}
+	if orphanCfg.Enabled {
+		log.Info("orphaned-runner sweep enabled", "grace", orphanCfg.Grace)
+	}
+
 	// Start scheduler (ties CI provider jobs to container lifecycle)
 	sched := scheduler.New(scheduler.Config{
 		Runtime:            rt,
@@ -626,6 +639,7 @@ func serve(ctx context.Context, configFile, imagesDirFlag string, containerdTCPP
 		NativeMacUser:      cfg.Runner.MacOS.User,
 		RunnerDir:          rm.Dir(),
 		Retry:              retryCfg,
+		OrphanSweep:        orphanCfg,
 		Log:                log,
 	})
 

config.example.toml

@@ -84,26 +84,21 @@ shutdown_timeout = "5m"
 # schedule = ["30s", "1m", "2m", "5m", "10m"]
 # jitter   = 0.2
 
-# Claim retry queue.
+# Orphaned-runner sweep.
 #
-# GitHub does not re-deliver workflow_job webhooks. When ephemerd's
-# initial claim/provision attempt fails for a transient reason
-# (rate-limit exhausted, transient 5xx, network blip), the job would
-# otherwise be lost. The retry queue schedules re-attempts on a
-# jittered exponential backoff ladder and gives up after max_age.
-#
-# When the last GitHub API response says remaining==0 with a known
-# reset time, the next attempt is snapped to just after reset instead
-# of falling through the ladder, so we don't burn attempts against a
-# provably-exhausted budget.
+# ephemerd registers one just-in-time runner per queued job, but GitHub
+# assigns a registered runner to ANY queued job with matching labels.
+# Runner teardown therefore follows the assignment GitHub reports in
+# workflow_job webhooks (runner_name), not the job the runner was
+# dispatched for. A runner whose intended job was picked up elsewhere
+# and that never received a job of its own is destroyed and
+# deregistered after the grace window.
 #
-# Enabled by default. Set enabled = false to restore the pre-existing
-# "log and drop" behavior.
-# [runner.claim_retry]
-# enabled  = true
-# max_age  = "90m"
-# schedule = ["30s", "1m", "2m", "5m", "10m"]
-# jitter   = 0.2
+# Only active in webhook mode (polling mode has no in_progress events
+# to observe assignments with). Enabled by default.
+# [runner.orphan_sweep]
+# enabled = true
+# grace   = "10m"
 
 [vm.linux]
 # Enable a Linux VM for running Linux jobs on Windows/macOS hosts.

pkg/config/config.go

@@ -469,6 +469,32 @@ type RunnerConfig struct {
 	// hit an API blip at claim time would be lost until human
 	// intervention.
 	ClaimRetry ClaimRetryToml `toml:"claim_retry"`
+
+	// OrphanSweep controls teardown of runners that were dispatched for
+	// a job but never observed picking one up. GitHub assigns JIT
+	// runners to ANY queued job with matching labels, so runner
+	// lifecycle is keyed to the observed assignment; a runner whose
+	// intended job went elsewhere and that never got a job of its own is
+	// destroyed after the grace window.
+	OrphanSweep OrphanSweepToml `toml:"orphan_sweep"`
+}
+
+// OrphanSweepToml configures the scheduler's orphaned-runner sweep.
+//
+// Enabled defaults to true when the [runner.orphan_sweep] table is
+// omitted. The sweep only acts in webhook mode (in polling mode there
+// are no in_progress events, so ephemerd cannot tell an orphaned runner
+// from a busy one) and only for providers that report runner
+// assignments (GitHub).
+type OrphanSweepToml struct {
+	// Enabled toggles the sweep. Nil = default true; operators disable
+	// by explicitly setting enabled = false.
+	Enabled *bool `toml:"enabled"`
+
+	// Grace is how long a dispatched runner may sit without being
+	// assigned a job before it is destroyed and deregistered. Accepts
+	// Go duration strings ("10m", "1h"). Default 10m.
+	Grace string `toml:"grace"`
 }
 
 // ClaimRetryToml configures the scheduler's claim-retry queue.
@@ -675,6 +701,29 @@ func (r *RunnerConfig) ClaimRetrySchedule() []time.Duration {
 	return out
 }
 
+// OrphanSweepEnabled reports whether the orphaned-runner sweep should
+// run. Defaults to true when the table is omitted or Enabled is nil;
+// operators opt out with `enabled = false`.
+func (r *RunnerConfig) OrphanSweepEnabled() bool {
+	if r == nil || r.OrphanSweep.Enabled == nil {
+		return true
+	}
+	return *r.OrphanSweep.Enabled
+}
+
+// OrphanSweepGrace returns how long a dispatched runner may remain
+// unassigned before the sweep destroys it. Defaults to 10 minutes when
+// unset or unparseable.
+func (r *RunnerConfig) OrphanSweepGrace() time.Duration {
+	if r == nil || r.OrphanSweep.Grace == "" {
+		return 10 * time.Minute
+	}
+	if d, err := time.ParseDuration(r.OrphanSweep.Grace); err == nil && d > 0 {
+		return d
+	}
+	return 10 * time.Minute
+}
+
 // ClaimRetryJitter returns the retry backoff jitter fraction. Defaults
 // to 0.2 (+/-20%). Values outside [0,1] are clamped to the default.
 func (r *RunnerConfig) ClaimRetryJitter() float64 {

pkg/config/config_test.go

@@ -2078,3 +2078,45 @@ func TestMacOSRunnerConfig_ResolvedMaxNative(t *testing.T) {
 		})
 	}
 }
+
+func TestOrphanSweepEnabled(t *testing.T) {
+	boolPtr := func(b bool) *bool { return &b }
+	tests := []struct {
+		name string
+		cfg  *RunnerConfig
+		want bool
+	}{
+		{"nil config defaults to true", nil, true},
+		{"omitted table defaults to true", &RunnerConfig{}, true},
+		{"explicit true", &RunnerConfig{OrphanSweep: OrphanSweepToml{Enabled: boolPtr(true)}}, true},
+		{"explicit false disables", &RunnerConfig{OrphanSweep: OrphanSweepToml{Enabled: boolPtr(false)}}, false},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := tt.cfg.OrphanSweepEnabled(); got != tt.want {
+				t.Errorf("OrphanSweepEnabled() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}
+
+func TestOrphanSweepGrace(t *testing.T) {
+	tests := []struct {
+		name string
+		cfg  *RunnerConfig
+		want time.Duration
+	}{
+		{"nil config defaults to 10m", nil, 10 * time.Minute},
+		{"empty defaults to 10m", &RunnerConfig{}, 10 * time.Minute},
+		{"custom duration", &RunnerConfig{OrphanSweep: OrphanSweepToml{Grace: "25m"}}, 25 * time.Minute},
+		{"unparseable defaults to 10m", &RunnerConfig{OrphanSweep: OrphanSweepToml{Grace: "soon"}}, 10 * time.Minute},
+		{"non-positive defaults to 10m", &RunnerConfig{OrphanSweep: OrphanSweepToml{Grace: "-5m"}}, 10 * time.Minute},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := tt.cfg.OrphanSweepGrace(); got != tt.want {
+				t.Errorf("OrphanSweepGrace() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}

pkg/github/client.go

@@ -505,6 +505,7 @@ func (c *Client) WebhookHandler(secret string) (http.Handler, <-chan JobEvent) {
 			"repo", repoName,
 			"job_id", payload.WorkflowJob.GetID(),
 			"labels", payload.WorkflowJob.Labels,
+			"runner_name", payload.WorkflowJob.GetRunnerName(),
 		)
 
 		events <- JobEvent{

pkg/providers/github/github.go

@@ -35,8 +35,9 @@ type Provider struct {
 
 // Compile-time interface checks.
 var (
-	_ providers.Poll    = (*Provider)(nil)
-	_ providers.Webhook = (*Provider)(nil)
+	_ providers.Poll               = (*Provider)(nil)
+	_ providers.Webhook            = (*Provider)(nil)
+	_ providers.RunnerNameReporter = (*Provider)(nil)
 )
 
 // New creates a GitHub provider wrapping an existing GitHub client.
@@ -231,6 +232,17 @@ func (p *Provider) convertEvent(ev github.JobEvent) providers.JobEvent {
 		fe.JobID = ev.Job.GetID()
 		fe.RunID = ev.Job.GetRunID()
 		fe.Conclusion = ev.Job.GetConclusion()
+		// GitHub populates runner_name on in_progress and completed
+		// actions; empty on queued and for jobs cancelled before any
+		// runner picked them up.
+		fe.RunnerName = ev.Job.GetRunnerName()
 	}
 	return fe
 }
+
+// ReportsRunnerNames implements providers.RunnerNameReporter: GitHub
+// workflow_job webhooks carry runner_name on in_progress and completed
+// actions, so the scheduler may key runner teardown and the orphan
+// sweep on observed assignments for runners dispatched via this
+// provider.
+func (p *Provider) ReportsRunnerNames() bool { return true }

pkg/providers/github/github_test.go

@@ -110,6 +110,46 @@ func TestConvertEvent_NilJob(t *testing.T) {
 	}
 }
 
+func TestConvertEvent_RunnerName(t *testing.T) {
+	p := New(nil, slog.Default(), "", "")
+
+	id := int64(86444761252)
+	runnerName := "ephemerd-github-ephpm-kind_cope"
+	job := &gh.WorkflowJob{
+		ID:         &id,
+		RunnerName: &runnerName,
+		Labels:     []string{"self-hosted", "linux"},
+	}
+
+	ev := p.convertEvent(ghclient.JobEvent{
+		Action: "in_progress",
+		Repo:   "ephpm/ephemerd",
+		Job:    job,
+	})
+
+	if ev.RunnerName != runnerName {
+		t.Errorf("RunnerName = %q, want %q", ev.RunnerName, runnerName)
+	}
+
+	// Queued events (and cancelled-before-assignment completions) carry
+	// no runner_name — the field must stay empty, not panic.
+	ev = p.convertEvent(ghclient.JobEvent{
+		Action: "queued",
+		Repo:   "ephpm/ephemerd",
+		Job:    &gh.WorkflowJob{ID: &id},
+	})
+	if ev.RunnerName != "" {
+		t.Errorf("RunnerName = %q for queued event, want empty", ev.RunnerName)
+	}
+}
+
+func TestReportsRunnerNames(t *testing.T) {
+	p := New(nil, slog.Default(), "", "")
+	if !p.ReportsRunnerNames() {
+		t.Error("ReportsRunnerNames() = false, want true — GitHub webhooks carry runner_name")
+	}
+}
+
 func TestStop_NilCancel(t *testing.T) {
 	p := New(nil, slog.Default(), "", "")
 	if err := p.Stop(context.Background()); err != nil {

pkg/providers/provider.go

@@ -109,6 +109,19 @@ type Webhook interface {
 	DeregisterWebhooks(ctx context.Context) error
 }
 
+// RunnerNameReporter is optionally implemented by providers whose
+// in_progress / completed JobEvents carry the assigned runner's name
+// (JobEvent.RunnerName). The scheduler only trusts "this runner was
+// never assigned a job" conclusions — e.g. the orphaned-runner sweep —
+// for runners dispatched via providers that report assignments.
+type RunnerNameReporter interface {
+	Provider
+
+	// ReportsRunnerNames returns true when the provider populates
+	// JobEvent.RunnerName on in_progress and completed events.
+	ReportsRunnerNames() bool
+}
+
 // PollConfig provides settings for poll-based job discovery.
 type PollConfig struct {
 	PollInterval int // seconds between polls (0 = provider default)
@@ -128,6 +141,17 @@ type JobEvent struct {
 	Labels     []string // runner labels/tags (e.g., ["self-hosted", "linux", "x64"])
 	Conclusion string   // for completed events: "success", "failure", "cancelled"
 
+	// RunnerName is the name of the runner the platform actually assigned
+	// the job to. GitHub populates it on "in_progress" and "completed"
+	// workflow_job webhook actions; it is empty on "queued" events, when a
+	// job was cancelled before any runner picked it up, and for providers
+	// that don't report it. The scheduler uses this to key runner teardown
+	// on the OBSERVED assignment rather than the dispatch intent — GitHub
+	// treats registered JIT runners with matching labels as fungible, so
+	// the runner ephemerd dispatched "for" a job is not necessarily the
+	// runner that ran it.
+	RunnerName string
+
 	// Raw holds the original forge-specific object for edge cases.
 	// GitHub: *github.WorkflowJob, Forgejo: task proto, GitLab: job payload.
 	Raw any