fix(scheduler): never destroy a runner that is executing a job

docs/arch/webhook-job-lifecycle.md

@@ -259,3 +259,119 @@ after the fact is strictly safe — it can waste a dispatch, never lose a job.
 
 A job over `maxProvisionAttempts` is not counted as class demand. We have given
 up on it, so it must not keep a spare runner alive indefinitely.
+
+## Third fix: never destroy a runner that is executing a job
+
+The two fixes above made the sweep's *rules* correct. They did not make its
+*evidence* correct, and that is what was killing live builds.
+
+Observed 2026-08-12 on the `linux-amd64` pool: three `dind-test` jobs with
+identical labels, dispatched concurrently, and a runner retired out from under
+an in-flight build.
+
+### Mechanism
+
+"Unbound" was inferred from webhooks and never verified. `runnerBinding.bound`
+flips only when `handleInProgress` processes an `in_progress` delivery naming
+that runner. Until then the runner looks idle in the ledger no matter what it
+is actually doing.
+
+That is not a rare, dropped-delivery failure — it is the normal shape of a
+same-label burst:
+
+1. Three same-label jobs A, B, C queue. ephemerd dispatches runners rA, rB, rC
+   (intent keys A, B, C). All three intent keys are now in `served`, so class
+   demand is **0**.
+2. GitHub permutes the assignments — say rA→B, rB→C, rC→A. Each runner starts
+   executing immediately.
+3. The three `in_progress` deliveries arrive over three separate HTTP requests
+   and are processed in whatever order they land. The **first** one processed
+   (job A, naming rC) sets `started[A]` and binds rC.
+4. `started[A]` is exactly the discharge signal for **rA**, whose intent key is
+   A. rA is now: unbound (its own `in_progress`, for job B, has not arrived),
+   discharged, and in a class with zero demand. That is rule 2, and rule 2
+   fires **immediately** — no grace window involved.
+5. Any sweep landing in that window retires rA. `handleCompleted` calls
+   `sweepOrphanRunners` on **every** completion, so it lands there routinely.
+
+rA was executing job B. The window is the inter-delivery skew between two
+`in_progress` webhooks of the same burst — hundreds of milliseconds is enough,
+and no delivery has to be dropped, delayed or reordered for it to open. Rule 1
+kills the same runner more slowly whenever a delivery genuinely *is* lost: the
+grace window expires and the runner still looks unbound.
+
+### Fix: the rules nominate, a busy check vetoes
+
+Both rules now produce **nominations**. Nothing is unhooked from the
+bookkeeping maps until a check taken *at the moment of teardown* — not derived
+from event history — confirms the runner is not executing a job. The hard
+invariant is: **never destroy a runner that is executing a job.**
+
+The check is layered (`pkg/scheduler/busy.go`, `pkg/runnerbusy`):
+
+1. **Local introspection (primary).** The actions-runner forks a
+   `Runner.Worker` child only while a job is executing — the listener process
+   is alive for the runner's whole life, so "a runner process exists" is not
+   the signal; the worker is. ephemerd owns the runtime, so it can look:
+   - **Linux (containerd):** `task.Pids()` → read `argv[0]` (falling back to
+     `comm`) from `/proc` on the host.
+   - **Windows (HCS / Hyper-V isolated):** the guest's processes are invisible
+     to the host process table, so HCS's own `ProcessList()` is used — the same
+     handle `pkg/metrics` already opens per container — matching `ImageName`.
+   - **macOS VM:** `pgrep -x Runner.Worker` over the existing per-job SSH
+     channel.
+   - **Native macOS:** `pgrep -g <pgid> -x Runner.Worker` against the runner's
+     process group on the host.
+   - **Dispatched into the Linux sidecar VM:** explicitly **unavailable** —
+     that containerd is behind the dispatch gRPC boundary and the host has no
+     view into the guest's PID namespace. It degrades to layer 2 rather than
+     silently answering "not busy".
+2. **The provider's busy flag (secondary).** GitHub reports `busy` per runner.
+   Consulted only when the local probe cannot answer, and only for a runner
+   already nominated — one GET per nomination, off every hot path.
+3. **Unknown.** Anything else. Unknown is *not* idle: it means possibly busy.
+
+The probes run with `s.mu` released (they do I/O), which opens a window for an
+`in_progress` delivery to bind a nominated runner. `reapRunnerLocked`
+re-validates the ledger entry — same binding, still unbound — before unhooking
+anything, so a runner bound during the probe survives its own nomination.
+
+### Escape hatches, and why both are time-based
+
+A veto that could never be overridden trades one leak for another: a wedged
+runner would squat a concurrency slot forever. Two bounds, both logged at warn
+level with an `ESCAPE:` prefix, and both counted in
+`ephemerd_orphan_reap_decisions_total{outcome="escaped"}`:
+
+- **Unknown → the grace window.** If the busy state cannot be determined for
+  the whole grace window, teardown proceeds. This is exactly the pre-veto
+  behaviour, so a platform with no usable probe is never *worse* than before —
+  and rule 1's nominations, which are already past the grace window, keep their
+  original timing on such a platform.
+- **Busy → the hard bound.** A positive busy verdict is overridden only past
+  `job_timeout + 30m` (or GitHub's 6h per-job ceiling + 30m when no job timeout
+  is configured). A job that exceeds `job_timeout` has already had its context
+  cancelled and its runner torn down by the normal path, so a runner still
+  claiming to be busy out there is wedged, not working.
+
+Both escapes are measured in **elapsed time**, not in consecutive failed
+probes. The sweep runs on every job completion as well as on a timer, so a
+probe-count escape would fire within milliseconds on a busy node — during
+exactly the same-label burst this change exists to survive — and effectively
+never on a quiet one. Wedged-ness is a property of duration, so duration is
+what bounds it.
+
+### Consequence: `orphan_grace` stops being load-bearing
+
+`orphan_grace` existed to paper over precisely this uncertainty. Too short
+killed live work; too long squatted a concurrency slot. With a verified-idle
+answer, reaping is immediate and safe, and with a verified-busy answer no
+window length can kill a build. The knob now only governs how long an
+*undeterminable* runner is held — the fallback path — so per-pool tuning
+(2m here, 15m there, 90m in the original incident) is no longer needed on any
+pool whose runners are locally probeable.
+
+The same veto answers the cross-daemon case: two pools both claiming arm64,
+GitHub binds one, and the loser's runner squats. "Is it actually busy?" is the
+right question there too, and the loser answers "no" immediately instead of
+waiting out a window.

pkg/github/client.go

@@ -256,6 +256,36 @@ func (c *Client) RemoveRunner(ctx context.Context, repo string, runnerID int64)
 	return nil
 }
 
+// RunnerBusy reports whether a self-hosted runner is currently executing
+// a job, from GitHub's own per-runner `busy` flag.
+//
+// Uses the org-level or repo-level API depending on configuration, the
+// same way RemoveRunner does — a JIT runner registered at the org level
+// is not addressable through the repo endpoint.
+//
+// A runner that has already deregistered itself (404) is reported as not
+// busy: an ephemeral runner removes itself after finishing its job, so
+// "gone" is the strongest possible evidence that nothing is running on
+// it. Every other error is returned to the caller, which must treat it as
+// "could not determine" rather than as idle.
+func (c *Client) RunnerBusy(ctx context.Context, repo string, runnerID int64) (bool, error) {
+	var runner *gh.Runner
+	var resp *gh.Response
+	var err error
+	if c.IsOrgLevel() {
+		runner, resp, err = c.client.Actions.GetOrganizationRunner(ctx, c.cfg.Owner, runnerID)
+	} else {
+		runner, resp, err = c.client.Actions.GetRunner(ctx, c.cfg.Owner, repo, runnerID)
+	}
+	if err != nil {
+		if resp != nil && resp.StatusCode == http.StatusNotFound {
+			return false, nil
+		}
+		return false, fmt.Errorf("reading runner %d: %w", runnerID, err)
+	}
+	return runner.GetBusy(), nil
+}
+
 // FetchJobImage fetches the workflow run's job definition and reads the
 // container image declared in the job's `container:` field. This requires an
 // extra API call per job but lets users specify the image directly in their

pkg/github/runnerbusy_test.go

@@ -0,0 +1,109 @@
+package github
+
+import (
+	"context"
+	"encoding/json"
+	"net/http"
+	"testing"
+)
+
+// TestRunnerBusy_RepoLevel pins the repo-scoped read of GitHub's own
+// per-runner busy flag. It is the scheduler's fallback authority for the
+// "never destroy a runner that is executing a job" invariant, used where
+// ephemerd cannot introspect the runner's runtime locally.
+func TestRunnerBusy_RepoLevel(t *testing.T) {
+	tests := []struct {
+		name string
+		busy bool
+	}{
+		{name: "runner is running a job", busy: true},
+		{name: "runner is idle", busy: false},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			mux := http.NewServeMux()
+			mux.HandleFunc("/repos/testorg/repo1/actions/runners/42", func(w http.ResponseWriter, r *http.Request) {
+				if err := json.NewEncoder(w).Encode(map[string]any{
+					"id":     42,
+					"name":   "runner-42",
+					"status": "online",
+					"busy":   tt.busy,
+				}); err != nil {
+					t.Logf("encoding: %v", err)
+				}
+			})
+			c, srv := newTestClientWithServer(t, mux)
+			defer srv.Close()
+
+			got, err := c.RunnerBusy(context.Background(), "repo1", 42)
+			if err != nil {
+				t.Fatalf("RunnerBusy: %v", err)
+			}
+			if got != tt.busy {
+				t.Errorf("RunnerBusy = %v, want %v", got, tt.busy)
+			}
+		})
+	}
+}
+
+// TestRunnerBusy_OrgLevel pins that an org-registered JIT runner is read
+// through the org endpoint. A repo-scoped GET would 404 for it, and a 404
+// reads as "not busy" — so getting the scope wrong would silently defeat
+// the veto for every org-level pool.
+func TestRunnerBusy_OrgLevel(t *testing.T) {
+	mux := http.NewServeMux()
+	mux.HandleFunc("/orgs/testorg/actions/runners/42", func(w http.ResponseWriter, r *http.Request) {
+		if err := json.NewEncoder(w).Encode(map[string]any{"id": 42, "busy": true}); err != nil {
+			t.Logf("encoding: %v", err)
+		}
+	})
+	c, srv := newTestClientWithServer(t, mux)
+	defer srv.Close()
+	c.cfg.Repos = nil // org-level
+
+	got, err := c.RunnerBusy(context.Background(), "", 42)
+	if err != nil {
+		t.Fatalf("RunnerBusy: %v", err)
+	}
+	if !got {
+		t.Error("RunnerBusy = false, want true")
+	}
+}
+
+// TestRunnerBusy_GoneIsNotBusy pins the one error that is an answer: an
+// ephemeral runner deregisters itself the moment its job finishes, so a
+// 404 is the strongest possible evidence that nothing is running on it.
+func TestRunnerBusy_GoneIsNotBusy(t *testing.T) {
+	mux := http.NewServeMux()
+	mux.HandleFunc("/repos/testorg/repo1/actions/runners/42", func(w http.ResponseWriter, r *http.Request) {
+		http.Error(w, `{"message":"Not Found"}`, http.StatusNotFound)
+	})
+	c, srv := newTestClientWithServer(t, mux)
+	defer srv.Close()
+
+	got, err := c.RunnerBusy(context.Background(), "repo1", 42)
+	if err != nil {
+		t.Fatalf("RunnerBusy: %v", err)
+	}
+	if got {
+		t.Error("a deregistered runner reported busy")
+	}
+}
+
+// TestRunnerBusy_ErrorIsNotAnAnswer pins the fail-safe contract at the
+// API boundary: anything other than a clean read or a 404 must surface as
+// an error, so the scheduler treats the runner as possibly busy rather
+// than as idle.
+func TestRunnerBusy_ErrorIsNotAnAnswer(t *testing.T) {
+	mux := http.NewServeMux()
+	mux.HandleFunc("/repos/testorg/repo1/actions/runners/42", func(w http.ResponseWriter, r *http.Request) {
+		http.Error(w, `{"message":"rate limit exceeded"}`, http.StatusForbidden)
+	})
+	c, srv := newTestClientWithServer(t, mux)
+	defer srv.Close()
+
+	if _, err := c.RunnerBusy(context.Background(), "repo1", 42); err == nil {
+		t.Fatal("want an error for a failed read; a rate limit must not read as idle")
+	}
+}

pkg/metrics/metrics.go

@@ -28,6 +28,16 @@ var (
 		Help: "Total number of jobs received (queued events).",
 	})
 
+	// OrphanReapDecisions counts what the orphan sweep did with each
+	// runner it nominated for teardown, by the busy verdict that decided
+	// it ("idle", "busy", "unknown") and the outcome ("reaped", "vetoed",
+	// "escaped"). A non-zero "escaped" rate means the busy check is
+	// failing to answer and runners are being destroyed on a timer again.
+	OrphanReapDecisions = promauto.NewCounterVec(prometheus.CounterOpts{
+		Name: "ephemerd_orphan_reap_decisions_total",
+		Help: "Orphan-sweep teardown nominations by busy verdict and outcome.",
+	}, []string{"verdict", "outcome"})
+
 	// JobDuration tracks the full lifecycle duration of a job.
 	JobDuration = promauto.NewHistogramVec(prometheus.HistogramOpts{
 		Name:    "ephemerd_job_duration_seconds",

pkg/native/native_darwin.go

@@ -17,6 +17,8 @@ import (
 	"strings"
 	"sync"
 	"syscall"
+
+	"github.com/ephpm/ephemerd/pkg/runnerbusy"
 )
 
 // serviceUserMu serializes service user creation across concurrent job starts.
@@ -406,6 +408,13 @@ func (r *Runner) Start(ctx context.Context) error {
 	return nil
 }
 
+// BusyState reports whether this runner is executing a job right now,
+// from the host's process table rather than from webhook history. The
+// scheduler uses it to veto teardown of a live job — see pkg/runnerbusy.
+func (r *Runner) BusyState(ctx context.Context) (runnerbusy.State, error) {
+	return runnerbusy.ProcessGroupBusy(ctx, r.pgid)
+}
+
 // Wait blocks until the runner process exits and returns its exit code.
 func (r *Runner) Wait() (int, error) {
 	if r.cmd == nil || r.cmd.Process == nil {

pkg/native/native_other.go

@@ -6,6 +6,8 @@ import (
 	"context"
 	"fmt"
 	"log/slog"
+
+	"github.com/ephpm/ephemerd/pkg/runnerbusy"
 )
 
 // Runner is a stub on non-darwin platforms.
@@ -37,3 +39,8 @@ func (r *Runner) Wait() (int, error) {
 
 // Stop is a stub on non-darwin platforms.
 func (r *Runner) Stop() {}
+
+// BusyState is a stub on non-darwin platforms.
+func (r *Runner) BusyState(ctx context.Context) (runnerbusy.State, error) {
+	return runnerbusy.ProcessGroupBusy(ctx, 0)
+}

pkg/providers/github/github.go

@@ -38,6 +38,7 @@ var (
 	_ providers.Poll               = (*Provider)(nil)
 	_ providers.Webhook            = (*Provider)(nil)
 	_ providers.RunnerNameReporter = (*Provider)(nil)
+	_ providers.RunnerBusyReporter = (*Provider)(nil)
 )
 
 // New creates a GitHub provider wrapping an existing GitHub client.
@@ -147,6 +148,23 @@ func (p *Provider) ReleaseJob(ctx context.Context, claim *providers.Claim) error
 	return p.client.RemoveRunner(ctx, claim.Repo, claim.RunnerID)
 }
 
+// RunnerBusy implements providers.RunnerBusyReporter. GitHub tracks a
+// per-runner `busy` flag on the Actions runners API and flips it for the
+// duration of an assignment, so it answers the teardown question — "is
+// something running on this runner" — without depending on a webhook
+// having been delivered.
+//
+// It is the scheduler's fallback authority, reached only when the local
+// probe cannot see the runner (a runner dispatched into the Linux sidecar
+// VM), and only for a runner already nominated for teardown. That is one
+// GET per nomination, not per job.
+func (p *Provider) RunnerBusy(ctx context.Context, claim *providers.Claim) (bool, error) {
+	if claim == nil {
+		return false, fmt.Errorf("nil claim")
+	}
+	return p.client.RunnerBusy(ctx, claim.Repo, claim.RunnerID)
+}
+
 func (p *Provider) FetchJobImage(ctx context.Context, event *providers.JobEvent) string {
 	return p.client.FetchJobImage(ctx, event.Repo, event.RunID, event.JobID)
 }

pkg/providers/provider.go

@@ -122,6 +122,27 @@ type RunnerNameReporter interface {
 	ReportsRunnerNames() bool
 }
 
+// RunnerBusyReporter is optionally implemented by providers that can be
+// asked, on demand, whether a runner they registered is executing a job
+// right now.
+//
+// This is the SECOND authority behind the scheduler's "never destroy a
+// busy runner" invariant. The first is a local probe of the container /
+// VM / process ephemerd owns, which needs no network and no API budget.
+// The provider is consulted only when the local probe cannot answer —
+// notably for runners dispatched into a sidecar VM, whose runtime lives
+// on the far side of a gRPC boundary. Because it is only reached at
+// teardown time, for a runner already nominated for reaping, it is off
+// every hot path.
+type RunnerBusyReporter interface {
+	Provider
+
+	// RunnerBusy reports whether the runner behind claim is currently
+	// running a job. An error means "could not determine" — callers must
+	// fail safe and treat that as possibly busy, never as idle.
+	RunnerBusy(ctx context.Context, claim *Claim) (bool, error)
+}
+
 // PollConfig provides settings for poll-based job discovery.
 type PollConfig struct {
 	PollInterval int // seconds between polls (0 = provider default)