diff --git a/docs/arch/webhook-job-lifecycle.md b/docs/arch/webhook-job-lifecycle.md
index 557aa63..c37d5fd 100644
--- a/docs/arch/webhook-job-lifecycle.md
+++ b/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.
diff --git a/pkg/github/client.go b/pkg/github/client.go
index b46a25d..820e681 100644
--- a/pkg/github/client.go
+++ b/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
diff --git a/pkg/github/runnerbusy_test.go b/pkg/github/runnerbusy_test.go
new file mode 100644
index 0000000..2f8d4cd
--- /dev/null
+++ b/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")
+	}
+}
diff --git a/pkg/metrics/metrics.go b/pkg/metrics/metrics.go
index 4d42895..731e130 100644
--- a/pkg/metrics/metrics.go
+++ b/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",
diff --git a/pkg/native/native_darwin.go b/pkg/native/native_darwin.go
index fd77dc5..0b6f706 100644
--- a/pkg/native/native_darwin.go
+++ b/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 {
diff --git a/pkg/native/native_other.go b/pkg/native/native_other.go
index 5ce3604..fbd0fd5 100644
--- a/pkg/native/native_other.go
+++ b/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)
+}
diff --git a/pkg/providers/github/github.go b/pkg/providers/github/github.go
index 9be3f18..601f956 100644
--- a/pkg/providers/github/github.go
+++ b/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)
 }
diff --git a/pkg/providers/provider.go b/pkg/providers/provider.go
index eedc708..4886614 100644
--- a/pkg/providers/provider.go
+++ b/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)
diff --git a/pkg/runnerbusy/container_linux.go b/pkg/runnerbusy/container_linux.go
new file mode 100644
index 0000000..ad3e3cd
--- /dev/null
+++ b/pkg/runnerbusy/container_linux.go
@@ -0,0 +1,76 @@
+//go:build linux
+
+package runnerbusy
+
+import (
+	"bytes"
+	"context"
+	"fmt"
+	"log/slog"
+	"os"
+	"strconv"
+)
+
+// ContainerBusy reports whether the runner inside a containerd-managed
+// Linux container is executing a job.
+//
+// runc leaves container processes visible in the host's PID namespace, so
+// the PIDs containerd reports are readable under /proc on the host. For
+// each one we read argv[0] (falling back to the kernel's comm) and look
+// for the runner's worker child.
+//
+// Failure modes all resolve to Unknown, never Idle:
+//
+//   - the task query fails (shim gone, containerd restarting)
+//   - the task reports no processes at all, which is indistinguishable
+//     from a listing that raced container teardown
+//   - every /proc read failed, which means we are not in a position to
+//     see the container's processes (foreign PID namespace, hardened
+//     /proc) rather than that the container is quiet
+func ContainerBusy(ctx context.Context, t ContainerTask, log *slog.Logger) (State, error) {
+	pids, err := t.Pids(ctx)
+	if err != nil {
+		return Unknown, fmt.Errorf("listing processes in container %s: %w", t.ID(), err)
+	}
+	if len(pids) == 0 {
+		return Unknown, fmt.Errorf("container %s reported no processes", t.ID())
+	}
+
+	read := 0
+	for _, p := range pids {
+		name, err := processName(int(p.Pid))
+		if err != nil {
+			// The process exited between the listing and the read. That
+			// is normal churn in a busy container, not a probe failure —
+			// keep going and judge on what we could read.
+			log.Debug("busy probe could not read a container process", "container", t.ID(), "pid", p.Pid, "error", err)
+			continue
+		}
+		read++
+		if IsWorkerProcess(name) {
+			return Busy, nil
+		}
+	}
+	if read == 0 {
+		return Unknown, fmt.Errorf("container %s: none of its %d processes were readable under /proc", t.ID(), len(pids))
+	}
+	return Idle, nil
+}
+
+// processName returns argv[0] for a host PID, falling back to the kernel's
+// comm when the cmdline is empty (kernel threads, or a process caught
+// mid-exec). "Runner.Worker" is 13 bytes, so it survives comm's 15-byte
+// truncation intact.
+func processName(pid int) (string, error) {
+	dir := "/proc/" + strconv.Itoa(pid)
+	if b, err := os.ReadFile(dir + "/cmdline"); err == nil {
+		if argv0, _, _ := bytes.Cut(b, []byte{0}); len(argv0) > 0 {
+			return string(argv0), nil
+		}
+	}
+	b, err := os.ReadFile(dir + "/comm")
+	if err != nil {
+		return "", fmt.Errorf("reading name of pid %d: %w", pid, err)
+	}
+	return string(bytes.TrimSpace(b)), nil
+}
diff --git a/pkg/runnerbusy/container_linux_test.go b/pkg/runnerbusy/container_linux_test.go
new file mode 100644
index 0000000..0ce48ce
--- /dev/null
+++ b/pkg/runnerbusy/container_linux_test.go
@@ -0,0 +1,152 @@
+//go:build linux
+
+package runnerbusy
+
+import (
+	"context"
+	"errors"
+	"io"
+	"log/slog"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"testing"
+
+	"github.com/containerd/containerd/v2/client"
+)
+
+type fakeTask struct {
+	id   string
+	pids []client.ProcessInfo
+	err  error
+}
+
+func (f *fakeTask) ID() string { return f.id }
+func (f *fakeTask) Pids(context.Context) ([]client.ProcessInfo, error) {
+	return f.pids, f.err
+}
+
+func quiet() *slog.Logger {
+	return slog.New(slog.NewTextHandler(io.Discard, nil))
+}
+
+// startFakeWorker execs a real process whose argv[0] basename is
+// Runner.Worker, so the probe reads it out of /proc exactly as it would
+// read the actions-runner's own worker. Returns its PID.
+func startFakeWorker(t *testing.T) uint32 {
+	t.Helper()
+	sleep, err := exec.LookPath("sleep")
+	if err != nil {
+		t.Skipf("no sleep binary to impersonate a worker: %v", err)
+	}
+	src, err := os.ReadFile(sleep)
+	if err != nil {
+		t.Skipf("cannot read %s: %v", sleep, err)
+	}
+	path := filepath.Join(t.TempDir(), "Runner.Worker")
+	if err := os.WriteFile(path, src, 0o755); err != nil {
+		t.Fatalf("writing fake worker: %v", err)
+	}
+	cmd := exec.Command(path, "60")
+	if err := cmd.Start(); err != nil {
+		t.Fatalf("starting fake worker: %v", err)
+	}
+	t.Cleanup(func() {
+		if err := cmd.Process.Kill(); err != nil {
+			t.Logf("killing fake worker: %v", err)
+		}
+		if _, err := cmd.Process.Wait(); err != nil {
+			t.Logf("reaping fake worker: %v", err)
+		}
+	})
+	return uint32(cmd.Process.Pid)
+}
+
+// TestContainerBusy_SeesAWorker is the Linux probe end to end: a real
+// process named Runner.Worker, found through a task's PID listing and
+// read out of /proc, must report Busy. This is the observation that
+// vetoes teardown of a live build.
+func TestContainerBusy_SeesAWorker(t *testing.T) {
+	worker := startFakeWorker(t)
+	task := &fakeTask{
+		id:   "job-1",
+		pids: []client.ProcessInfo{{Pid: uint32(os.Getpid())}, {Pid: worker}},
+	}
+	got, err := ContainerBusy(context.Background(), task, quiet())
+	if err != nil {
+		t.Fatalf("ContainerBusy: %v", err)
+	}
+	if got != Busy {
+		t.Errorf("state = %v, want busy", got)
+	}
+}
+
+// TestContainerBusy_NoWorkerIsIdle pins the other half: a readable
+// process listing with no worker in it is a POSITIVE idle observation,
+// which is what makes immediate reaping safe.
+func TestContainerBusy_NoWorkerIsIdle(t *testing.T) {
+	task := &fakeTask{
+		id:   "job-1",
+		pids: []client.ProcessInfo{{Pid: uint32(os.Getpid())}},
+	}
+	got, err := ContainerBusy(context.Background(), task, quiet())
+	if err != nil {
+		t.Fatalf("ContainerBusy: %v", err)
+	}
+	if got != Idle {
+		t.Errorf("state = %v, want idle", got)
+	}
+}
+
+// TestContainerBusy_FailuresAreUnknown pins the fail-safe contract: every
+// way the probe can fail to see the container's processes reports
+// Unknown, so the caller vetoes teardown instead of assuming idle.
+func TestContainerBusy_FailuresAreUnknown(t *testing.T) {
+	tests := []struct {
+		name string
+		task *fakeTask
+	}{
+		{
+			name: "task query failed",
+			task: &fakeTask{id: "job-1", err: errors.New("shim gone")},
+		},
+		{
+			name: "task reports no processes at all",
+			task: &fakeTask{id: "job-1"},
+		},
+		{
+			// PIDs we cannot read under /proc mean we are not in a
+			// position to see this container, not that it is quiet.
+			name: "no listed process was readable",
+			task: &fakeTask{id: "job-1", pids: []client.ProcessInfo{{Pid: 1 << 30}, {Pid: 1<<30 + 1}}},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got, err := ContainerBusy(context.Background(), tt.task, quiet())
+			if got != Unknown {
+				t.Errorf("state = %v, want unknown", got)
+			}
+			if err == nil {
+				t.Error("want an error explaining why the probe could not answer")
+			}
+		})
+	}
+}
+
+// TestProcessName_FallsBackToComm pins the fallback used when a process
+// has no readable cmdline. "Runner.Worker" is 13 bytes, so it survives
+// comm's truncation and still matches.
+func TestProcessName_FallsBackToComm(t *testing.T) {
+	name, err := processName(os.Getpid())
+	if err != nil {
+		t.Fatalf("processName(self): %v", err)
+	}
+	if name == "" {
+		t.Error("processName(self) is empty")
+	}
+	if _, err := processName(1 << 30); err == nil {
+		t.Error("processName of a nonexistent pid should fail, not return a name")
+	}
+}
diff --git a/pkg/runnerbusy/container_other.go b/pkg/runnerbusy/container_other.go
new file mode 100644
index 0000000..35bb068
--- /dev/null
+++ b/pkg/runnerbusy/container_other.go
@@ -0,0 +1,22 @@
+//go:build !linux && !windows
+
+package runnerbusy
+
+import (
+	"context"
+	"log/slog"
+)
+
+// ContainerBusy has no implementation on this platform.
+//
+// The only non-Linux, non-Windows host ephemerd runs on is macOS, and a
+// macOS host never owns a runner container directly: Linux jobs there are
+// dispatched into the Linux sidecar VM, whose containerd lives on the far
+// side of a gRPC boundary, and macOS jobs run in a per-job macOS VM
+// probed over SSH instead (see the scheduler's macOS VM prober).
+//
+// This returns Unknown + ErrUnsupported so the caller falls through to
+// the next authority rather than mistaking "no probe" for "not busy".
+func ContainerBusy(_ context.Context, _ ContainerTask, _ *slog.Logger) (State, error) {
+	return Unknown, ErrUnsupported
+}
diff --git a/pkg/runnerbusy/container_windows.go b/pkg/runnerbusy/container_windows.go
new file mode 100644
index 0000000..5da6eb1
--- /dev/null
+++ b/pkg/runnerbusy/container_windows.go
@@ -0,0 +1,71 @@
+//go:build windows
+
+package runnerbusy
+
+import (
+	"context"
+	"fmt"
+	"log/slog"
+
+	"github.com/Microsoft/hcsshim"
+)
+
+// hcsProcessLister is the minimal subset of hcsshim.Container the Windows
+// probe needs. Injectable so the probe can be tested without a live
+// compute system.
+type hcsProcessLister interface {
+	ProcessList() ([]hcsshim.ProcessListItem, error)
+	Close() error
+}
+
+// openCompute is swapped out in tests.
+var openCompute = func(id string) (hcsProcessLister, error) {
+	c, err := hcsshim.OpenContainer(id)
+	if err != nil {
+		return nil, err
+	}
+	return c, nil
+}
+
+// ContainerBusy reports whether the runner inside a Windows container is
+// executing a job.
+//
+// Windows runner containers are Hyper-V isolated, so the runner's
+// processes live in a utility VM and are NOT visible in the host's
+// process table — the Linux probe's /proc walk has no host-side
+// equivalent here, and inspecting the host would silently report "no
+// worker" for every job. The Host Compute Service is the supported way
+// in: it proxies a process listing out of the guest through the GCS, for
+// both process-isolated and Hyper-V isolated containers, and reports each
+// process's image name. This is the same handle pkg/metrics already opens
+// per container for resource sampling.
+//
+// Same fail-safe rule as the Linux probe: anything we cannot read is
+// Unknown, and an empty listing (which HCS also returns for a compute
+// system that has gone away underneath us) is Unknown rather than Idle.
+func ContainerBusy(_ context.Context, t ContainerTask, log *slog.Logger) (State, error) {
+	id := t.ID()
+	c, err := openCompute(id)
+	if err != nil {
+		return Unknown, fmt.Errorf("opening compute system %q: %w", id, err)
+	}
+	defer func() {
+		if err := c.Close(); err != nil {
+			log.Debug("closing compute system after busy probe", "container", id, "error", err)
+		}
+	}()
+
+	list, err := c.ProcessList()
+	if err != nil {
+		return Unknown, fmt.Errorf("listing processes in compute system %q: %w", id, err)
+	}
+	if len(list) == 0 {
+		return Unknown, fmt.Errorf("compute system %q reported no processes", id)
+	}
+	for _, p := range list {
+		if IsWorkerProcess(p.ImageName) {
+			return Busy, nil
+		}
+	}
+	return Idle, nil
+}
diff --git a/pkg/runnerbusy/container_windows_test.go b/pkg/runnerbusy/container_windows_test.go
new file mode 100644
index 0000000..3c7678b
--- /dev/null
+++ b/pkg/runnerbusy/container_windows_test.go
@@ -0,0 +1,148 @@
+//go:build windows
+
+package runnerbusy
+
+import (
+	"context"
+	"errors"
+	"io"
+	"log/slog"
+	"testing"
+
+	"github.com/Microsoft/hcsshim"
+	"github.com/containerd/containerd/v2/client"
+)
+
+type fakeTask struct{ id string }
+
+func (f *fakeTask) ID() string { return f.id }
+func (f *fakeTask) Pids(context.Context) ([]client.ProcessInfo, error) {
+	return nil, errors.New("the windows probe must not need containerd PIDs")
+}
+
+type fakeCompute struct {
+	list     []hcsshim.ProcessListItem
+	listErr  error
+	closeErr error
+	closed   bool
+}
+
+func (f *fakeCompute) ProcessList() ([]hcsshim.ProcessListItem, error) {
+	return f.list, f.listErr
+}
+
+func (f *fakeCompute) Close() error {
+	f.closed = true
+	return f.closeErr
+}
+
+func quiet() *slog.Logger {
+	return slog.New(slog.NewTextHandler(io.Discard, nil))
+}
+
+func withCompute(t *testing.T, c hcsProcessLister, err error) {
+	t.Helper()
+	prev := openCompute
+	openCompute = func(string) (hcsProcessLister, error) {
+		if err != nil {
+			return nil, err
+		}
+		return c, nil
+	}
+	t.Cleanup(func() { openCompute = prev })
+}
+
+// TestContainerBusy_Windows pins the HCS probe. Hyper-V isolated
+// containers keep their processes in a utility VM, invisible to the host
+// process table, so the compute system's own process list — reported
+// through the GCS with an ImageName per process — is the only ground
+// truth available.
+func TestContainerBusy_Windows(t *testing.T) {
+	tests := []struct {
+		name     string
+		list     []hcsshim.ProcessListItem
+		listErr  error
+		openErr  error
+		want     State
+		wantErr  bool
+		wantOpen bool
+	}{
+		{
+			name: "worker present means busy",
+			list: []hcsshim.ProcessListItem{
+				{ProcessId: 1, ImageName: "Runner.Listener.exe"},
+				{ProcessId: 2, ImageName: "Runner.Worker.exe"},
+			},
+			want:     Busy,
+			wantOpen: true,
+		},
+		{
+			// The listener alone is what an IDLE runner looks like: it is
+			// alive for the runner's whole life. Treating "a runner
+			// process exists" as busy would veto every teardown forever.
+			name: "listener alone means idle",
+			list: []hcsshim.ProcessListItem{
+				{ProcessId: 1, ImageName: "Runner.Listener.exe"},
+				{ProcessId: 3, ImageName: "cmd.exe"},
+			},
+			want:     Idle,
+			wantOpen: true,
+		},
+		{
+			name:     "empty process list is unknown, not idle",
+			want:     Unknown,
+			wantErr:  true,
+			wantOpen: true,
+		},
+		{
+			name:     "process list error is unknown",
+			listErr:  errors.New("compute system is shutting down"),
+			want:     Unknown,
+			wantErr:  true,
+			wantOpen: true,
+		},
+		{
+			name:    "cannot open the compute system",
+			openErr: errors.New("no such compute system"),
+			want:    Unknown,
+			wantErr: true,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			fake := &fakeCompute{list: tt.list, listErr: tt.listErr}
+			withCompute(t, fake, tt.openErr)
+
+			got, err := ContainerBusy(context.Background(), &fakeTask{id: "job-1"}, quiet())
+			if got != tt.want {
+				t.Errorf("state = %v, want %v", got, tt.want)
+			}
+			if (err != nil) != tt.wantErr {
+				t.Errorf("err = %v, wantErr = %v", err, tt.wantErr)
+			}
+			if fake.closed != tt.wantOpen {
+				t.Errorf("compute system closed = %v, want %v", fake.closed, tt.wantOpen)
+			}
+		})
+	}
+}
+
+// TestContainerBusy_WindowsCloseErrorIsNotFatal pins that a failure to
+// release the HCS handle does not change the verdict — the answer is
+// already computed and the handle is process-local.
+func TestContainerBusy_WindowsCloseErrorIsNotFatal(t *testing.T) {
+	fake := &fakeCompute{
+		list:     []hcsshim.ProcessListItem{{ProcessId: 2, ImageName: "Runner.Worker.exe"}},
+		closeErr: errors.New("handle already gone"),
+	}
+	withCompute(t, fake, nil)
+
+	got, err := ContainerBusy(context.Background(), &fakeTask{id: "job-1"}, quiet())
+	if err != nil {
+		t.Fatalf("ContainerBusy: %v", err)
+	}
+	if got != Busy {
+		t.Errorf("state = %v, want busy", got)
+	}
+}
diff --git a/pkg/runnerbusy/procgroup_darwin.go b/pkg/runnerbusy/procgroup_darwin.go
new file mode 100644
index 0000000..a0ade57
--- /dev/null
+++ b/pkg/runnerbusy/procgroup_darwin.go
@@ -0,0 +1,45 @@
+//go:build darwin
+
+package runnerbusy
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"os/exec"
+	"strconv"
+)
+
+// ProcessGroupBusy reports whether a native macOS runner — one that runs
+// as a plain host process group rather than inside a container or VM — is
+// executing a job.
+//
+// The runner is started as a process-group leader (pgid == the listener's
+// pid), and every process it forks, including the per-job worker, inherits
+// that group. So "does this process group contain a Runner.Worker" is the
+// same question the container probes ask, expressed in the units macOS
+// gives us.
+//
+// pgrep exit codes: 0 = matched, 1 = no match, anything else = the query
+// itself failed. Only the first two are answers; the rest is Unknown.
+func ProcessGroupBusy(ctx context.Context, pgid int) (State, error) {
+	if pgid <= 0 {
+		return Unknown, errors.New("native runner has no process group yet")
+	}
+	// -g restricts the search to the runner's process group; -x demands an
+	// exact process-name match so a step that happens to mention the
+	// worker in its argv cannot fake a busy verdict.
+	cmd := exec.CommandContext(ctx, "pgrep", "-g", strconv.Itoa(pgid), "-x", "Runner.Worker")
+	out, err := cmd.Output()
+	if err == nil {
+		if len(out) == 0 {
+			return Unknown, errors.New("pgrep matched but returned no pids")
+		}
+		return Busy, nil
+	}
+	var exitErr *exec.ExitError
+	if errors.As(err, &exitErr) && exitErr.ExitCode() == 1 {
+		return Idle, nil
+	}
+	return Unknown, fmt.Errorf("pgrep for a worker in process group %d: %w", pgid, err)
+}
diff --git a/pkg/runnerbusy/procgroup_other.go b/pkg/runnerbusy/procgroup_other.go
new file mode 100644
index 0000000..8707a90
--- /dev/null
+++ b/pkg/runnerbusy/procgroup_other.go
@@ -0,0 +1,13 @@
+//go:build !darwin
+
+package runnerbusy
+
+import "context"
+
+// ProcessGroupBusy has no implementation off macOS: native (un-contained)
+// runners only exist on macOS hosts. Returns Unknown + ErrUnsupported so
+// the caller falls through rather than mistaking "no probe" for "not
+// busy".
+func ProcessGroupBusy(_ context.Context, _ int) (State, error) {
+	return Unknown, ErrUnsupported
+}
diff --git a/pkg/runnerbusy/runnerbusy.go b/pkg/runnerbusy/runnerbusy.go
new file mode 100644
index 0000000..d59d2fa
--- /dev/null
+++ b/pkg/runnerbusy/runnerbusy.go
@@ -0,0 +1,105 @@
+// Package runnerbusy answers one question from ground truth rather than
+// inference: is a GitHub Actions runner executing a job RIGHT NOW?
+//
+// ephemerd otherwise only learns that a runner is busy by processing an
+// `in_progress` webhook. That is inference ABOUT busy-ness, not an
+// observation of it: anything that delays, drops or reorders a delivery —
+// and a burst of same-label jobs makes all three likely — leaves a runner
+// that is genuinely executing a build looking idle in the scheduler's
+// ledger. Teardown decisions taken on that belief kill live builds.
+//
+// The signal used here is the actions-runner's own process model. The
+// listener process (`Runner.Listener`) runs for the whole life of the
+// runner, so "a runner process exists" says nothing. The listener forks a
+// `Runner.Worker` child ONLY while a job is executing, and reaps it when
+// the job ends. "A worker exists" is therefore equivalent to "a job is
+// running", and ephemerd — which owns the container, VM or process the
+// runner lives in — can observe it locally: no network, no API budget,
+// and immune to a missed or reordered webhook.
+//
+// Every probe returns a State. Unknown is NOT idle. Callers MUST treat
+// Unknown as "possibly busy" and fail safe; a probe that cannot answer
+// never reports Idle.
+package runnerbusy
+
+import (
+	"context"
+	"errors"
+	"strings"
+
+	"github.com/containerd/containerd/v2/client"
+)
+
+// State is a probe's verdict about a runner.
+type State int
+
+const (
+	// Unknown means the probe could not determine the runner's state:
+	// the platform has no local probe, the runtime refused the query, or
+	// the answer was empty in a way that cannot be distinguished from a
+	// failure. Callers must treat it as "possibly busy".
+	Unknown State = iota
+
+	// Idle means the probe positively observed that no job is executing:
+	// the runner's process list was read successfully and contained no
+	// worker. Only this verdict makes teardown safe.
+	Idle
+
+	// Busy means the probe positively observed a worker process, i.e. a
+	// job is executing right now. Destroying the runner would kill it.
+	Busy
+)
+
+func (s State) String() string {
+	switch s {
+	case Idle:
+		return "idle"
+	case Busy:
+		return "busy"
+	default:
+		return "unknown"
+	}
+}
+
+// ErrUnsupported is returned by probes that have no implementation on the
+// running platform. It is an explicit "I cannot answer" — never an
+// implicit "not busy".
+var ErrUnsupported = errors.New("runnerbusy: no local probe on this platform")
+
+// workerProcess is the actions-runner's per-job child process. It exists
+// for exactly as long as a job is executing.
+const workerProcess = "runner.worker"
+
+// IsWorkerProcess reports whether a process identifier names the
+// actions-runner's job worker.
+//
+// The argument is whatever the platform's process listing yields for a
+// process: argv[0] on Linux (an absolute path such as
+// /home/runner/bin/Runner.Worker), /proc/<pid>/comm as a fallback, or the
+// HCS ImageName on Windows (Runner.Worker.exe). Matching is on the base
+// name, case-insensitively, with a .exe suffix stripped, so one rule
+// covers all three.
+//
+// Runner.Listener is deliberately NOT matched: it is alive for the whole
+// life of the runner, including while the runner sits idle waiting for a
+// job, so matching it would make every runner look permanently busy.
+func IsWorkerProcess(name string) bool {
+	name = strings.TrimSpace(name)
+	if i := strings.LastIndexAny(name, `/\`); i >= 0 {
+		name = name[i+1:]
+	}
+	name = strings.ToLower(name)
+	name = strings.TrimSuffix(name, ".exe")
+	return name == workerProcess
+}
+
+// ContainerTask is the slice of containerd's client.Task that the
+// container probe needs. Declared as an interface so the probe can be
+// unit-tested without a containerd daemon.
+//
+// ID returns the container ID (the init task's ID is the container's).
+// Pids lists the processes in the container.
+type ContainerTask interface {
+	ID() string
+	Pids(context.Context) ([]client.ProcessInfo, error)
+}
diff --git a/pkg/runnerbusy/runnerbusy_test.go b/pkg/runnerbusy/runnerbusy_test.go
new file mode 100644
index 0000000..04abfda
--- /dev/null
+++ b/pkg/runnerbusy/runnerbusy_test.go
@@ -0,0 +1,69 @@
+package runnerbusy
+
+import "testing"
+
+// TestIsWorkerProcess pins the one rule that decides "a job is running
+// here". It has to cover three different shapes of process identifier —
+// Linux argv[0], Linux comm, Windows HCS ImageName — and it must NOT
+// match the listener, which is alive for the whole life of an idle runner
+// and would make every runner look permanently busy.
+func TestIsWorkerProcess(t *testing.T) {
+	tests := []struct {
+		name string
+		in   string
+		want bool
+	}{
+		{"linux argv0, official image", "/home/runner/bin/Runner.Worker", true},
+		{"linux argv0, mounted runner", "/actions-runner/bin/Runner.Worker", true},
+		{"linux comm", "Runner.Worker", true},
+		{"windows hcs image name", "Runner.Worker.exe", true},
+		{"windows full path", `C:\actions-runner\bin\Runner.Worker.exe`, true},
+		{"case insensitive", "runner.worker.exe", true},
+		{"surrounding whitespace", "  Runner.Worker\n", true},
+
+		{"listener is not a worker", "/home/runner/bin/Runner.Listener", false},
+		{"listener exe is not a worker", "Runner.Listener.exe", false},
+		{"run.sh is not a worker", "/home/runner/run.sh", false},
+		{"a job step is not a worker", "/usr/bin/bash", false},
+		{"substring does not match", "Runner.WorkerHelper", false},
+		{"prefix does not match", "My.Runner.Worker", false},
+		{"empty", "", false},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := IsWorkerProcess(tt.in); got != tt.want {
+				t.Errorf("IsWorkerProcess(%q) = %v, want %v", tt.in, got, tt.want)
+			}
+		})
+	}
+}
+
+// TestStateString pins the strings that end up in log lines and in the
+// orphan-sweep metric's verdict label.
+func TestStateString(t *testing.T) {
+	tests := []struct {
+		in   State
+		want string
+	}{
+		{Unknown, "unknown"},
+		{Idle, "idle"},
+		{Busy, "busy"},
+		{State(42), "unknown"},
+	}
+	for _, tt := range tests {
+		if got := tt.in.String(); got != tt.want {
+			t.Errorf("State(%d).String() = %q, want %q", tt.in, got, tt.want)
+		}
+	}
+}
+
+// TestZeroValueIsUnknown pins the fail-safe default: a State nobody set
+// must read as "could not determine", never as idle. Every probe returns
+// the zero value on its error paths.
+func TestZeroValueIsUnknown(t *testing.T) {
+	var s State
+	if s != Unknown {
+		t.Fatalf("zero State = %v, want Unknown — an unset verdict must never read as idle", s)
+	}
+}
diff --git a/pkg/scheduler/busy.go b/pkg/scheduler/busy.go
new file mode 100644
index 0000000..ebb81a0
--- /dev/null
+++ b/pkg/scheduler/busy.go
@@ -0,0 +1,346 @@
+package scheduler
+
+import (
+	"context"
+	"time"
+
+	"github.com/containerd/containerd/v2/pkg/namespaces"
+	"github.com/ephpm/ephemerd/pkg/metrics"
+	"github.com/ephpm/ephemerd/pkg/providers"
+	"github.com/ephpm/ephemerd/pkg/runnerbusy"
+	"github.com/ephpm/ephemerd/pkg/runtime"
+)
+
+// The hard invariant this file enforces: NEVER DESTROY A RUNNER THAT IS
+// EXECUTING A JOB.
+//
+// The orphan sweep's two rules (intent-keyed grace, label-set fungibility
+// reconciliation) are good at spotting runners that have lost their
+// purpose, but both decide from the same evidence: the scheduler's belief
+// about which runner is bound to which job, assembled from in_progress
+// webhooks. That belief is inference, and a same-label burst is exactly
+// the load that makes it wrong — three JIT runners are dispatched, GitHub
+// permutes the assignments, and between the first in_progress delivery
+// and the last there is a window in which a runner that is already
+// executing a build still looks unbound. A sweep landing in that window
+// (every completed event triggers one) retires a live job.
+//
+// So the rules no longer decide. They NOMINATE, and a busy check taken at
+// the moment of teardown either confirms the nomination or vetoes it. The
+// check asks the runtime ephemerd owns — is there a Runner.Worker process
+// in there? — and falls back to GitHub's per-runner busy flag when it
+// cannot see the runner at all. Anything short of a positive "idle"
+// answer is a veto.
+//
+// A veto that could never be overridden would trade one leak for
+// another — a wedged runner would squat a concurrency slot forever — so
+// the veto is bounded in time. See reapPolicy and decideReap.
+
+const (
+	// busyProbeTimeout bounds a single busy check. The local probes are
+	// sub-millisecond; the bound exists so a wedged containerd shim or a
+	// stalled GitHub request cannot hold the sweep (and, through it, the
+	// scheduler mutex on the next phase) open.
+	busyProbeTimeout = 10 * time.Second
+
+	// maxJobRuntime is the longest a single job can legitimately run when
+	// no job_timeout is configured — GitHub's own per-job ceiling for
+	// self-hosted runners. Used to derive the hard bound below.
+	maxJobRuntime = 6 * time.Hour
+
+	// hardBoundMargin is how far past a job's own deadline a runner may
+	// keep vetoing its own teardown before we call it wedged. A job that
+	// exceeds JobTimeout has already had its context cancelled and its
+	// runner torn down by the normal path, so a runner still claiming to
+	// be busy this far out is not running anything we can save.
+	hardBoundMargin = 30 * time.Minute
+)
+
+// reapPolicy is the pair of time bounds that keep the busy veto from
+// becoming permanent.
+type reapPolicy struct {
+	// Grace is the orphan sweep's existing grace window. It doubles as
+	// the ceiling on an UNKNOWN verdict: if we cannot determine whether a
+	// runner is busy, we fall back to exactly the pre-veto behaviour
+	// (destroy once it has been unbound this long) rather than holding
+	// the slot indefinitely.
+	Grace time.Duration
+
+	// HardBound is the ceiling on a POSITIVE busy verdict, measured from
+	// dispatch. Past it, teardown proceeds over the veto with a loud log
+	// line: the runner is wedged, not working.
+	HardBound time.Duration
+}
+
+// reapPolicy derives the veto's time bounds from the scheduler config.
+func (s *Scheduler) reapPolicy() reapPolicy {
+	grace := s.cfg.OrphanSweep.Grace
+	if grace <= 0 {
+		grace = defaultOrphanGrace
+	}
+	limit := s.cfg.JobTimeout
+	if limit <= 0 {
+		limit = maxJobRuntime
+	}
+	hard := limit + hardBoundMargin
+	if hard < grace {
+		// A grace window longer than the hard bound would let the weaker
+		// (unknown) escape outlive the stronger (busy) one.
+		hard = grace
+	}
+	return reapPolicy{Grace: grace, HardBound: hard}
+}
+
+// reapInput is everything decideReap needs. Kept free of scheduler state
+// so the decision is a pure function of observable facts.
+type reapInput struct {
+	Runner       string
+	DispatchedAt time.Time
+	Now          time.Time
+	Busy         runnerbusy.State
+	Policy       reapPolicy
+}
+
+// Outcomes, also used as the `outcome` metric label.
+const (
+	outcomeReaped  = "reaped"
+	outcomeVetoed  = "vetoed"
+	outcomeEscaped = "escaped"
+)
+
+// reapDecision is decideReap's verdict on one nominated runner.
+type reapDecision struct {
+	// Destroy is whether teardown may proceed.
+	Destroy bool
+
+	// Escape is true when teardown proceeds DESPITE the busy check
+	// failing to clear the runner. Always logged at warn level: it is the
+	// only path by which this code can still kill a live job, and its
+	// rate is the signal that the busy check has stopped working.
+	Escape bool
+
+	// Outcome is one of outcomeReaped / outcomeVetoed / outcomeEscaped.
+	Outcome string
+
+	// Reason is a short, stable, human-readable explanation for the log.
+	Reason string
+}
+
+// decideReap turns a nomination plus a busy verdict into a teardown
+// decision. Pure: same inputs, same answer, no clock and no I/O.
+//
+// The three verdicts carry different weight, and the escape hatches are
+// sized accordingly.
+//
+//   - Idle is a positive observation that no job is executing. Teardown
+//     proceeds immediately — this is what makes aggressive reaping safe
+//     and what lets orphan_grace stop being load-bearing.
+//
+//   - Busy is a positive observation that a job IS executing. It vetoes
+//     teardown up to HardBound (job deadline plus a margin). Nothing
+//     legitimate is still running past that, so a runner that keeps
+//     answering "busy" there is wedged and gets destroyed.
+//
+//   - Unknown means we could not determine either way — no probe on this
+//     platform, a runtime that refused the query, an unreachable API. It
+//     is treated as possibly-busy and vetoes teardown, but only up to
+//     Grace, at which point we are back to the pre-veto behaviour: a
+//     runner unbound for the whole grace window is destroyed.
+//
+// Why both escapes are measured in TIME rather than in consecutive failed
+// probes: the sweep runs on every job completion as well as on a timer,
+// so a probe-count escape fires within milliseconds on a busy node during
+// exactly the same-label burst this whole change exists to survive, and
+// effectively never on a quiet one. Wedged-ness is a property of
+// duration, so duration is what bounds it.
+func decideReap(in reapInput) reapDecision {
+	age := in.Now.Sub(in.DispatchedAt)
+
+	if age >= in.Policy.HardBound && in.Busy != runnerbusy.Idle {
+		return reapDecision{
+			Destroy: true,
+			Escape:  true,
+			Outcome: outcomeEscaped,
+			Reason:  "wedged: still not idle past the hard bound",
+		}
+	}
+
+	switch in.Busy {
+	case runnerbusy.Idle:
+		return reapDecision{
+			Destroy: true,
+			Outcome: outcomeReaped,
+			Reason:  "verified idle: no worker process on the runner",
+		}
+	case runnerbusy.Busy:
+		return reapDecision{
+			Outcome: outcomeVetoed,
+			Reason:  "vetoed: the runner is executing a job",
+		}
+	default:
+		if age >= in.Policy.Grace {
+			return reapDecision{
+				Destroy: true,
+				Escape:  true,
+				Outcome: outcomeEscaped,
+				Reason:  "busy state undeterminable for the whole grace window",
+			}
+		}
+		return reapDecision{
+			Outcome: outcomeVetoed,
+			Reason:  "vetoed: busy state could not be determined",
+		}
+	}
+}
+
+// vetoBusyNominations is the gate in front of teardown.
+//
+// It takes the sweep's nominations, asks each one's runtime whether a job
+// is executing on it, and returns only the ones cleared for destruction —
+// already unhooked from the bookkeeping maps.
+//
+// Runs with s.mu RELEASED: the probes do I/O. That opens a window in
+// which an in_progress webhook can arrive and bind a nominated runner, so
+// reapRunnerLocked re-validates the ledger entry (same binding, still
+// unbound) before unhooking anything. A runner that got bound in the
+// window is silently kept, which is the correct answer.
+func (s *Scheduler) vetoBusyNominations(noms []orphanNomination, policy reapPolicy) []orphanVictim {
+	if len(noms) == 0 {
+		return nil
+	}
+	ctx := context.Background()
+	victims := make([]orphanVictim, 0, len(noms))
+
+	for _, n := range noms {
+		state, source := s.runnerBusy(ctx, n.rj)
+		d := decideReap(reapInput{
+			Runner:       n.name,
+			DispatchedAt: n.rb.dispatchedAt,
+			Now:          time.Now(),
+			Busy:         state,
+			Policy:       policy,
+		})
+		metrics.OrphanReapDecisions.WithLabelValues(state.String(), d.Outcome).Inc()
+
+		if !d.Destroy {
+			s.cfg.Log.Info("orphan sweep nomination vetoed",
+				"runner", n.name,
+				"dispatched_for_job", n.rb.intentKey.JobID,
+				"busy_verdict", state.String(),
+				"probe", source,
+				"reason", d.Reason)
+			continue
+		}
+
+		s.mu.Lock()
+		v, ok := s.reapRunnerLocked(n.name, n.rb)
+		s.mu.Unlock()
+		if !ok {
+			s.cfg.Log.Debug("orphan sweep nomination went stale before teardown",
+				"runner", n.name, "detail", "bound or cleaned up while the busy check ran")
+			continue
+		}
+		v.discharged = n.discharged
+		v.escaped = d.Escape
+		v.verdict = state.String()
+		v.reason = d.Reason
+		victims = append(victims, v)
+	}
+	return victims
+}
+
+// busyProber answers "is this runner executing a job right now?".
+// Swapped out in tests; the production implementation is
+// (*Scheduler).probeRunnerBusy.
+type busyProber func(ctx context.Context, rj *runningJob) (runnerbusy.State, string)
+
+// runnerBusy runs the configured prober, defaulting to the real one.
+func (s *Scheduler) runnerBusy(ctx context.Context, rj *runningJob) (runnerbusy.State, string) {
+	if s.busyProbe != nil {
+		return s.busyProbe(ctx, rj)
+	}
+	return s.probeRunnerBusy(ctx, rj)
+}
+
+// probeRunnerBusy is the layered busy check.
+//
+//  1. Local introspection of the container / VM / process ephemerd owns.
+//     Preferred: it is ground truth, needs no network, spends no API
+//     budget, and no webhook delivery can defeat it.
+//  2. The provider's own busy flag, for the paths where ephemerd cannot
+//     see into the runner (currently: runners dispatched into the Linux
+//     sidecar VM, whose containerd is behind the dispatch gRPC boundary).
+//  3. Unknown — which the caller must treat as possibly busy.
+//
+// The second return value names the authority that answered, for logs.
+func (s *Scheduler) probeRunnerBusy(ctx context.Context, rj *runningJob) (runnerbusy.State, string) {
+	ctx, cancel := context.WithTimeout(ctx, busyProbeTimeout)
+	defer cancel()
+
+	state, source, err := s.probeLocalBusy(ctx, rj)
+	if state != runnerbusy.Unknown {
+		return state, source
+	}
+	if err != nil {
+		s.cfg.Log.Debug("local busy probe could not answer",
+			"runner", rj.runnerName(), "probe", source, "error", err)
+	}
+
+	reporter, ok := rj.provider.(providers.RunnerBusyReporter)
+	if !ok || rj.claim == nil {
+		return runnerbusy.Unknown, source
+	}
+	busy, err := reporter.RunnerBusy(ctx, rj.claim)
+	if err != nil {
+		s.cfg.Log.Warn("provider busy check failed; treating the runner as possibly busy",
+			"runner", rj.runnerName(), "error", err)
+		return runnerbusy.Unknown, source
+	}
+	if busy {
+		return runnerbusy.Busy, "provider"
+	}
+	return runnerbusy.Idle, "provider"
+}
+
+// probeLocalBusy inspects whatever ephemerd itself owns for this job.
+//
+// Every branch either answers from a live process listing or returns
+// Unknown with an explicit reason. None of them can return Idle by
+// default, which is the property that makes the veto safe.
+func (s *Scheduler) probeLocalBusy(ctx context.Context, rj *runningJob) (runnerbusy.State, string, error) {
+	switch {
+	case rj.env != nil && rj.env.Task != nil:
+		// Linux: containerd task PIDs are host PIDs, read via /proc.
+		// Windows: HCS proxies a process listing out of the Hyper-V
+		// isolated container. Both live in pkg/runnerbusy behind build
+		// tags; macOS hosts have no local container and get Unknown.
+		st, err := runnerbusy.ContainerBusy(
+			namespaces.WithNamespace(ctx, runtime.Namespace), rj.env.Task, s.cfg.Log)
+		return st, "container", err
+
+	case rj.macosVM != nil:
+		st, err := s.macOSVMBusy(ctx, rj.macosVM)
+		return st, "macos-vm", err
+
+	case rj.nativeRunner != nil:
+		p, ok := rj.nativeRunner.(interface {
+			BusyState(context.Context) (runnerbusy.State, error)
+		})
+		if !ok {
+			return runnerbusy.Unknown, "native-process", runnerbusy.ErrUnsupported
+		}
+		st, err := p.BusyState(ctx)
+		return st, "native-process", err
+
+	case rj.dispatched != "":
+		// The runner is a container inside the Linux sidecar VM. Its
+		// containerd is reachable only through the dispatch gRPC service,
+		// which has no process-introspection call, and the host has no
+		// view into the guest's PID namespace. Explicitly unavailable
+		// rather than silently idle: the provider busy flag is the
+		// authority for this path.
+		return runnerbusy.Unknown, "dispatched", runnerbusy.ErrUnsupported
+	}
+
+	return runnerbusy.Unknown, "none", runnerbusy.ErrUnsupported
+}
diff --git a/pkg/scheduler/busy_macosvm.go b/pkg/scheduler/busy_macosvm.go
new file mode 100644
index 0000000..aa6890e
--- /dev/null
+++ b/pkg/scheduler/busy_macosvm.go
@@ -0,0 +1,102 @@
+package scheduler
+
+import (
+	"context"
+	"crypto/ed25519"
+	"errors"
+	"fmt"
+	"io"
+	"net"
+	"time"
+
+	"github.com/ephpm/ephemerd/pkg/runnerbusy"
+	"github.com/ephpm/ephemerd/pkg/vm"
+	"golang.org/x/crypto/ssh"
+)
+
+// macOSVMBusy reports whether the runner inside a per-job macOS VM is
+// executing a job.
+//
+// A macOS job runs the actions-runner as an ordinary process inside a
+// guest VM, so neither the /proc walk (Linux containers) nor HCS (Windows
+// containers) applies: from the host, a Virtualization.framework guest is
+// one opaque process. The guest is however already reachable over SSH on
+// the ephemeral key ephemerd generates at startup — that is how the job
+// is set up and how `ephemerd ssh` attaches — so the same channel answers
+// the process question. pgrep's exit status is the whole answer; nothing
+// is parsed out of the guest's stdout.
+//
+// Every failure to reach or interrogate the guest is Unknown, never Idle.
+func (s *Scheduler) macOSVMBusy(ctx context.Context, mv vm.MacOSVM) (runnerbusy.State, error) {
+	ip := mv.RunnerAddress()
+	if ip == "" {
+		return runnerbusy.Unknown, errors.New("macOS VM has no discovered address yet")
+	}
+
+	s.mu.Lock()
+	cfg := s.cfg.MacOSVMConfig
+	s.mu.Unlock()
+	if cfg == nil {
+		return runnerbusy.Unknown, errors.New("macOS VM config is not set")
+	}
+
+	var auth []ssh.AuthMethod
+	if key, ok := cfg.SSHSigner.(ed25519.PrivateKey); ok {
+		signer, err := ssh.NewSignerFromKey(key)
+		if err != nil {
+			return runnerbusy.Unknown, fmt.Errorf("building SSH signer for the guest: %w", err)
+		}
+		auth = append(auth, ssh.PublicKeys(signer))
+	}
+	if len(auth) == 0 {
+		return runnerbusy.Unknown, errors.New("no SSH key available for the guest")
+	}
+
+	deadline, ok := ctx.Deadline()
+	timeout := 5 * time.Second
+	if ok {
+		if d := time.Until(deadline); d > 0 && d < timeout {
+			timeout = d
+		}
+	}
+
+	// The guest is an ephemeral, per-job VM on a host-only network whose
+	// host key is regenerated with the clone, so there is no key to pin —
+	// same trust model the rest of the macOS VM plumbing uses.
+	client, err := ssh.Dial("tcp", net.JoinHostPort(ip, "22"), &ssh.ClientConfig{
+		User:            "admin",
+		Auth:            auth,
+		HostKeyCallback: ssh.InsecureIgnoreHostKey(),
+		Timeout:         timeout,
+	})
+	if err != nil {
+		return runnerbusy.Unknown, fmt.Errorf("connecting to the macOS guest at %s: %w", ip, err)
+	}
+	defer func() {
+		if err := client.Close(); err != nil {
+			s.cfg.Log.Debug("closing guest SSH session after busy probe", "ip", ip, "error", err)
+		}
+	}()
+
+	session, err := client.NewSession()
+	if err != nil {
+		return runnerbusy.Unknown, fmt.Errorf("opening a session on the macOS guest at %s: %w", ip, err)
+	}
+	defer func() {
+		if err := session.Close(); err != nil && !errors.Is(err, io.EOF) {
+			s.cfg.Log.Debug("closing guest SSH session after busy probe", "ip", ip, "error", err)
+		}
+	}()
+
+	// -x demands an exact process-name match, so a build step that merely
+	// mentions the worker in its command line cannot fake a busy verdict.
+	err = session.Run("pgrep -x Runner.Worker >/dev/null")
+	if err == nil {
+		return runnerbusy.Busy, nil
+	}
+	var exitErr *ssh.ExitError
+	if errors.As(err, &exitErr) && exitErr.ExitStatus() == 1 {
+		return runnerbusy.Idle, nil
+	}
+	return runnerbusy.Unknown, fmt.Errorf("probing for a worker on the macOS guest at %s: %w", ip, err)
+}
diff --git a/pkg/scheduler/busy_test.go b/pkg/scheduler/busy_test.go
new file mode 100644
index 0000000..1a764f4
--- /dev/null
+++ b/pkg/scheduler/busy_test.go
@@ -0,0 +1,465 @@
+package scheduler
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+
+	"github.com/ephpm/ephemerd/pkg/providers"
+	"github.com/ephpm/ephemerd/pkg/runnerbusy"
+)
+
+// Stand-ins for the real ground-truth check. Tests have no containerd,
+// no guest VM and no runner process to introspect, so the probe is
+// injected; what is under test here is what the scheduler DOES with each
+// answer.
+func idleProbe(context.Context, *runningJob) (runnerbusy.State, string) {
+	return runnerbusy.Idle, "test"
+}
+
+func busyProbeFn(context.Context, *runningJob) (runnerbusy.State, string) {
+	return runnerbusy.Busy, "test"
+}
+
+func unknownProbe(context.Context, *runningJob) (runnerbusy.State, string) {
+	return runnerbusy.Unknown, "test"
+}
+
+// TestDecideReap is the veto's decision matrix as a pure function:
+// nomination + busy verdict + elapsed time in, teardown decision out.
+//
+// The invariant it pins is the one this whole change exists for — a
+// runner observed to be executing a job is never destroyed — together
+// with the two bounded escapes that keep the veto from leaking a wedged
+// runner forever.
+func TestDecideReap(t *testing.T) {
+	const (
+		grace = 10 * time.Minute
+		hard  = 2 * time.Hour
+	)
+	policy := reapPolicy{Grace: grace, HardBound: hard}
+	now := time.Now()
+
+	tests := []struct {
+		name        string
+		busy        runnerbusy.State
+		age         time.Duration
+		wantDestroy bool
+		wantEscape  bool
+		wantOutcome string
+	}{
+		{
+			// The bug. A runner mid-build looks unbound because its
+			// in_progress webhook has not been processed yet; the rules
+			// nominate it; the check vetoes.
+			name:        "busy runner inside the grace window is never destroyed",
+			busy:        runnerbusy.Busy,
+			age:         time.Minute,
+			wantOutcome: outcomeVetoed,
+		},
+		{
+			// Rule 1's own window is no longer authority either: a
+			// runner still executing a job at the end of the grace
+			// window keeps running.
+			name:        "busy runner past the grace window is still not destroyed",
+			busy:        runnerbusy.Busy,
+			age:         grace + time.Minute,
+			wantOutcome: outcomeVetoed,
+		},
+		{
+			// What makes aggressive reaping safe: a positive idle
+			// observation retires the runner immediately, without
+			// waiting out any window.
+			name:        "verified-idle runner is retired immediately",
+			busy:        runnerbusy.Idle,
+			age:         time.Second,
+			wantDestroy: true,
+			wantOutcome: outcomeReaped,
+		},
+		{
+			name:        "verified-idle runner is retired past the hard bound too, without escaping",
+			busy:        runnerbusy.Idle,
+			age:         hard + time.Hour,
+			wantDestroy: true,
+			wantOutcome: outcomeReaped,
+		},
+		{
+			// Fail-safe: a probe that cannot answer must not read as idle.
+			name:        "undeterminable state inside the grace window is treated as busy",
+			busy:        runnerbusy.Unknown,
+			age:         time.Minute,
+			wantOutcome: outcomeVetoed,
+		},
+		{
+			// Escape 1: with no usable check we are back to exactly the
+			// pre-veto behaviour rather than holding the slot forever.
+			name:        "undeterminable state for the whole grace window escapes",
+			busy:        runnerbusy.Unknown,
+			age:         grace,
+			wantDestroy: true,
+			wantEscape:  true,
+			wantOutcome: outcomeEscaped,
+		},
+		{
+			// Escape 2: the only thing that overrides a POSITIVE busy
+			// verdict. No legitimate job is still running this far past
+			// its own deadline.
+			name:        "busy runner past the hard bound escapes",
+			busy:        runnerbusy.Busy,
+			age:         hard,
+			wantDestroy: true,
+			wantEscape:  true,
+			wantOutcome: outcomeEscaped,
+		},
+		{
+			name:        "unknown runner past the hard bound escapes",
+			busy:        runnerbusy.Unknown,
+			age:         hard + time.Minute,
+			wantDestroy: true,
+			wantEscape:  true,
+			wantOutcome: outcomeEscaped,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got := decideReap(reapInput{
+				Runner:       "r",
+				DispatchedAt: now.Add(-tt.age),
+				Now:          now,
+				Busy:         tt.busy,
+				Policy:       policy,
+			})
+			if got.Destroy != tt.wantDestroy {
+				t.Errorf("Destroy = %v, want %v (reason: %s)", got.Destroy, tt.wantDestroy, got.Reason)
+			}
+			if got.Escape != tt.wantEscape {
+				t.Errorf("Escape = %v, want %v (reason: %s)", got.Escape, tt.wantEscape, got.Reason)
+			}
+			if got.Outcome != tt.wantOutcome {
+				t.Errorf("Outcome = %q, want %q", got.Outcome, tt.wantOutcome)
+			}
+			if got.Reason == "" {
+				t.Error("Reason is empty; every decision must be explainable in the log")
+			}
+		})
+	}
+}
+
+// TestReapPolicy pins how the veto's two bounds are derived: the grace
+// window comes from config (with the package default as fallback), and
+// the hard bound sits a margin past the job deadline — GitHub's own
+// per-job ceiling when no deadline is configured.
+func TestReapPolicy(t *testing.T) {
+	tests := []struct {
+		name          string
+		grace         time.Duration
+		jobTimeout    time.Duration
+		wantGrace     time.Duration
+		wantHardBound time.Duration
+	}{
+		{
+			name:          "defaults",
+			wantGrace:     defaultOrphanGrace,
+			wantHardBound: maxJobRuntime + hardBoundMargin,
+		},
+		{
+			name:          "configured grace and job timeout",
+			grace:         2 * time.Minute,
+			jobTimeout:    90 * time.Minute,
+			wantGrace:     2 * time.Minute,
+			wantHardBound: 90*time.Minute + hardBoundMargin,
+		},
+		{
+			// A grace window longer than the derived hard bound would
+			// let the weak (unknown) escape outlive the strong (busy)
+			// one, which would be nonsense.
+			name:          "hard bound never sits below the grace window",
+			grace:         12 * time.Hour,
+			jobTimeout:    time.Minute,
+			wantGrace:     12 * time.Hour,
+			wantHardBound: 12 * time.Hour,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			s := New(Config{
+				OrphanSweep: OrphanSweepConfig{Enabled: true, Grace: tt.grace},
+				JobTimeout:  tt.jobTimeout,
+				Log:         quietLogger(),
+			})
+			got := s.reapPolicy()
+			if got.Grace != tt.wantGrace {
+				t.Errorf("Grace = %v, want %v", got.Grace, tt.wantGrace)
+			}
+			if got.HardBound != tt.wantHardBound {
+				t.Errorf("HardBound = %v, want %v", got.HardBound, tt.wantHardBound)
+			}
+		})
+	}
+}
+
+// TestSweepOrphanRunners_BusyVeto is the end-to-end statement of the
+// invariant, through the real sweep: the rules nominate a runner, the
+// busy check answers, and only a verified-idle runner is destroyed.
+//
+// Every row uses a runner that BOTH rules would retire without the veto:
+// discharged (its dispatch-intent job was observed running elsewhere),
+// no same-label demand left, and past the grace window.
+func TestSweepOrphanRunners_BusyVeto(t *testing.T) {
+	tests := []struct {
+		name        string
+		probe       busyProber
+		age         time.Duration
+		jobTimeout  time.Duration
+		wantDestroy bool
+	}{
+		{
+			name:        "busy runner is never destroyed",
+			probe:       busyProbeFn,
+			age:         30 * time.Minute,
+			jobTimeout:  time.Hour,
+			wantDestroy: false,
+		},
+		{
+			name:        "unavailable busy check is treated as busy inside the grace window",
+			probe:       unknownProbe,
+			age:         time.Minute,
+			jobTimeout:  time.Hour,
+			wantDestroy: false,
+		},
+		{
+			name:        "idle runner is destroyed",
+			probe:       idleProbe,
+			age:         30 * time.Minute,
+			jobTimeout:  time.Hour,
+			wantDestroy: true,
+		},
+		{
+			// Escape hatch: a runner that has claimed to be busy well
+			// past any job it could legitimately still be running is
+			// wedged, and must not squat a concurrency slot forever.
+			name:        "busy runner past the hard bound is destroyed anyway",
+			probe:       busyProbeFn,
+			age:         2 * time.Hour,
+			jobTimeout:  time.Minute,
+			wantDestroy: true,
+		},
+		{
+			name:        "unavailable busy check past the grace window is destroyed anyway",
+			probe:       unknownProbe,
+			age:         30 * time.Minute,
+			jobTimeout:  time.Hour,
+			wantDestroy: true,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			fake := &fakeDispatchServer{}
+			_, dc, stopDispatch := startFakeDispatchServer(t, fake)
+			defer stopDispatch()
+
+			base := newClaimCountingProvider("github")
+			prov := &reportingProvider{claimCountingProvider: base}
+
+			s := New(Config{
+				Providers:       []providers.Provider{prov},
+				LinuxDispatcher: dc,
+				JobTimeout:      tt.jobTimeout,
+				OrphanSweep:     OrphanSweepConfig{Enabled: true, Grace: 10 * time.Minute},
+				Log:             quietLogger(),
+			})
+			s.webhookMode = true
+			s.busyProbe = tt.probe
+
+			key := seedDispatchedRunner(s, prov, 1, "r-1", time.Now().Add(-tt.age), false, true)
+			s.mu.Lock()
+			s.seen[key] = time.Now()
+			s.jobLabels[key] = labelSetKey([]string{"self-hosted", "linux"})
+			s.runners["r-1"].labelSet = labelSetKey([]string{"self-hosted", "linux"})
+			// Discharged: the job it was dispatched for was observed
+			// running somewhere else, so rule 2 nominates it too.
+			s.started[key] = time.Now()
+			s.mu.Unlock()
+
+			s.sweepOrphanRunners()
+
+			if tt.wantDestroy {
+				waitForDestroy(t, fake, "r-1")
+			}
+			destroyed := destroyedNames(fake)["r-1"]
+			if destroyed != tt.wantDestroy {
+				t.Errorf("runner destroyed = %v, want %v", destroyed, tt.wantDestroy)
+			}
+
+			s.mu.Lock()
+			_, stillRunning := s.running[key]
+			_, stillLedgered := s.runners["r-1"]
+			s.mu.Unlock()
+			if stillRunning == tt.wantDestroy {
+				t.Errorf("running entry present = %v, want %v", stillRunning, !tt.wantDestroy)
+			}
+			if stillLedgered == tt.wantDestroy {
+				t.Errorf("ledger entry present = %v, want %v", stillLedgered, !tt.wantDestroy)
+			}
+		})
+	}
+}
+
+// TestSweepOrphanRunners_BindingRaceDuringProbe pins the reason the veto
+// runs with the scheduler mutex released and re-validates afterwards.
+//
+// The probe does I/O, so an in_progress webhook can land while it is in
+// flight — which is precisely the delivery skew that produced the
+// original bug. A runner that gets bound in that window must survive its
+// own nomination even if the probe said "idle" a moment earlier.
+func TestSweepOrphanRunners_BindingRaceDuringProbe(t *testing.T) {
+	fake := &fakeDispatchServer{}
+	_, dc, stopDispatch := startFakeDispatchServer(t, fake)
+	defer stopDispatch()
+
+	base := newClaimCountingProvider("github")
+	prov := &reportingProvider{claimCountingProvider: base}
+
+	s := New(Config{
+		Providers:       []providers.Provider{prov},
+		LinuxDispatcher: dc,
+		OrphanSweep:     OrphanSweepConfig{Enabled: true, Grace: 10 * time.Minute},
+		Log:             quietLogger(),
+	})
+	s.webhookMode = true
+
+	key := seedDispatchedRunner(s, prov, 1, "r-1", time.Now().Add(-30*time.Minute), false, true)
+
+	// The probe answers "idle", then the webhook arrives: exactly the
+	// interleaving the nominate/veto split has to survive.
+	s.busyProbe = func(context.Context, *runningJob) (runnerbusy.State, string) {
+		s.handleInProgress(providers.JobEvent{
+			Provider:   prov,
+			Action:     "in_progress",
+			JobID:      777,
+			Repo:       "myrepo",
+			RunnerName: "r-1",
+		})
+		return runnerbusy.Idle, "test"
+	}
+
+	s.sweepOrphanRunners()
+	time.Sleep(50 * time.Millisecond)
+
+	if destroyedNames(fake)["r-1"] {
+		t.Error("runner bound while the busy check was in flight was destroyed anyway")
+	}
+	s.mu.Lock()
+	_, stillRunning := s.running[key]
+	s.mu.Unlock()
+	if !stillRunning {
+		t.Error("runner bound during the probe was dropped from running")
+	}
+}
+
+// TestProbeLocalBusy_UnavailablePathsAreUnknown pins the fail-safe
+// contract on the local probe: every shape it cannot introspect reports
+// Unknown with an explicit reason. None may fall through to Idle, which
+// would be indistinguishable from "verified not running a job".
+func TestProbeLocalBusy_UnavailablePathsAreUnknown(t *testing.T) {
+	s := New(Config{Log: quietLogger()})
+
+	tests := []struct {
+		name       string
+		rj         *runningJob
+		wantSource string
+	}{
+		{
+			// Runner lives in the Linux sidecar VM, behind the dispatch
+			// gRPC boundary. The provider busy flag covers this path.
+			name:       "dispatched to the linux sidecar VM",
+			rj:         &runningJob{dispatched: "runner-x"},
+			wantSource: "dispatched",
+		},
+		{
+			name:       "native runner without a busy hook",
+			rj:         &runningJob{nativeRunner: stopOnly{}},
+			wantSource: "native-process",
+		},
+		{
+			name:       "nothing to introspect",
+			rj:         &runningJob{},
+			wantSource: "none",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			state, source, err := s.probeLocalBusy(context.Background(), tt.rj)
+			if state != runnerbusy.Unknown {
+				t.Errorf("state = %v, want unknown — an unavailable probe must never read as idle", state)
+			}
+			if source != tt.wantSource {
+				t.Errorf("source = %q, want %q", source, tt.wantSource)
+			}
+			if !errors.Is(err, runnerbusy.ErrUnsupported) {
+				t.Errorf("err = %v, want ErrUnsupported", err)
+			}
+		})
+	}
+}
+
+type stopOnly struct{}
+
+func (stopOnly) Stop() {}
+
+// TestProbeRunnerBusy_FallsBackToProvider pins the second layer: when the
+// local probe cannot see the runner, the provider's own busy flag decides,
+// and a provider error still fails safe to Unknown.
+func TestProbeRunnerBusy_FallsBackToProvider(t *testing.T) {
+	tests := []struct {
+		name  string
+		busy  bool
+		err   error
+		want  runnerbusy.State
+		wants string
+	}{
+		{name: "provider says busy", busy: true, want: runnerbusy.Busy, wants: "provider"},
+		{name: "provider says idle", want: runnerbusy.Idle, wants: "provider"},
+		{name: "provider errors", err: errors.New("rate limited"), want: runnerbusy.Unknown, wants: "dispatched"},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			s := New(Config{Log: quietLogger()})
+			prov := &busyReportingProvider{
+				claimCountingProvider: newClaimCountingProvider("github"),
+				busy:                  tt.busy,
+				err:                   tt.err,
+			}
+			rj := &runningJob{
+				dispatched: "runner-x",
+				provider:   prov,
+				claim:      &providers.Claim{RunnerID: 1, RunnerName: "runner-x", Repo: "myrepo"},
+			}
+			state, source := s.probeRunnerBusy(context.Background(), rj)
+			if state != tt.want {
+				t.Errorf("state = %v, want %v", state, tt.want)
+			}
+			if source != tt.wants {
+				t.Errorf("source = %q, want %q", source, tt.wants)
+			}
+		})
+	}
+}
+
+// busyReportingProvider implements providers.RunnerBusyReporter over the
+// shared test provider.
+type busyReportingProvider struct {
+	*claimCountingProvider
+	busy bool
+	err  error
+}
+
+func (p *busyReportingProvider) RunnerBusy(context.Context, *providers.Claim) (bool, error) {
+	return p.busy, p.err
+}
diff --git a/pkg/scheduler/fungibility_test.go b/pkg/scheduler/fungibility_test.go
index 6594c9c..a54adea 100644
--- a/pkg/scheduler/fungibility_test.go
+++ b/pkg/scheduler/fungibility_test.go
@@ -357,6 +357,12 @@ func TestSweepOrphanRunners_LabelSetReconciliation(t *testing.T) {
 				Log:             quietLogger(),
 			})
 			s.webhookMode = true
+			// This matrix pins which runners the rules NOMINATE. The
+			// busy check is a separate layer with its own tests
+			// (busy_test.go), so pin it to a verified-idle answer: a
+			// nomination the check cannot clear is vetoed, and every
+			// row here would otherwise be testing the veto instead.
+			s.busyProbe = idleProbe
 
 			keys := map[string]jobKey{}
 			for _, sp := range tt.spares {
diff --git a/pkg/scheduler/scheduler.go b/pkg/scheduler/scheduler.go
index 3d2b797..d4335a4 100644
--- a/pkg/scheduler/scheduler.go
+++ b/pkg/scheduler/scheduler.go
@@ -223,6 +223,13 @@ type Scheduler struct {
 	// retry holds pending re-attempts for jobs whose initial claim
 	// failed with a retryable error. Nil when Config.Retry.Enabled=false.
 	retry *retryQueue
+
+	// busyProbe overrides the ground-truth "is this runner executing a
+	// job" check that vetoes orphan teardown. Nil (the default) uses
+	// probeRunnerBusy: local container/VM/process introspection, falling
+	// back to the provider's busy flag. Set only by tests, which have no
+	// real runtime to introspect.
+	busyProbe busyProber
 }
 
 const seenTTL = 10 * time.Minute
@@ -1988,12 +1995,62 @@ type orphanVictim struct {
 	key        jobKey
 	rj         *runningJob
 	discharged bool
+
+	// escaped records that teardown proceeded over a busy veto that never
+	// cleared. Logged loudly — this is the only remaining path by which
+	// the sweep can kill a live job.
+	escaped bool
+	// verdict / reason carry the busy check's answer into the log line.
+	verdict string
+	reason  string
+}
+
+// orphanNomination is a runner the sweep's rules have PROPOSED for
+// teardown. It is not yet a victim: the bookkeeping maps still hold it,
+// and the busy check gets a veto before anything is unhooked.
+//
+// Nominating and reaping are deliberately separate phases because the
+// busy check does I/O (a containerd query, an SSH round trip, a GitHub
+// GET) and must not run under s.mu — and because in the window while it
+// runs, an in_progress webhook may arrive and bind the runner, which the
+// reap phase re-checks.
+type orphanNomination struct {
+	name       string
+	rb         *runnerBinding
+	rj         *runningJob
+	discharged bool
+}
+
+// nominateRunnerLocked resolves a ledger entry to the job it belongs to
+// without unhooking anything. Caller holds s.mu. Returns ok=false when the
+// entry is stale or no longer names this runner.
+func (s *Scheduler) nominateRunnerLocked(name string, rb *runnerBinding) (orphanNomination, bool) {
+	rj, ok := s.running[rb.intentKey]
+	if !ok {
+		// Stale ledger entry — the wait-goroutine already cleaned up.
+		delete(s.runners, name)
+		return orphanNomination{}, false
+	}
+	if rj.runnerName() != name {
+		return orphanNomination{}, false
+	}
+	return orphanNomination{name: name, rb: rb, rj: rj}, true
 }
 
 // reapRunnerLocked unhooks a runner from the bookkeeping maps and returns it
 // for teardown. Caller holds s.mu. Returns ok=false when the ledger entry is
-// stale or no longer names this runner, in which case nothing is torn down.
+// stale, no longer names this runner, or has been bound to a job since it was
+// nominated — the last case is the race the nominate/veto split exists to
+// catch: an in_progress webhook landing while the busy check was in flight.
 func (s *Scheduler) reapRunnerLocked(name string, rb *runnerBinding) (orphanVictim, bool) {
+	if cur, ok := s.runners[name]; !ok || cur != rb {
+		// The ledger entry was replaced or removed while the busy check
+		// ran; whatever is there now was not what we nominated.
+		return orphanVictim{}, false
+	}
+	if rb.bound {
+		return orphanVictim{}, false
+	}
 	rj, ok := s.running[rb.intentKey]
 	if !ok {
 		// Stale ledger entry — the wait-goroutine already cleaned up.
@@ -2040,16 +2097,30 @@ func (s *Scheduler) reapRunnerLocked(name string, rb *runnerBinding) (orphanVict
 // that class's uncovered queued jobs first, so a spare that GitHub can still
 // legitimately hand a sibling job is kept (and no second runner is dispatched
 // for that job), while genuine surplus is retired immediately.
+//
+// # Both rules only NOMINATE
+//
+// Both rules decide from the same evidence — the scheduler's belief about
+// which runner is bound, assembled from in_progress webhooks — and that
+// belief is inference, not observation. A same-label burst permutes three
+// JIT runners across three jobs, and between the first in_progress
+// delivery and the last there is a window in which a runner that is
+// already executing a build still looks unbound; every completed event
+// triggers a sweep, so landing in that window is routine. Reaping on that
+// belief killed live builds.
+//
+// So the rules produce NOMINATIONS, and a busy check taken at the moment
+// of teardown — not derived from event history — either confirms or
+// vetoes each one. See busy.go for the check and its (time-bounded)
+// escape hatches.
 func (s *Scheduler) sweepOrphanRunners() {
 	if !s.cfg.OrphanSweep.Enabled {
 		return
 	}
-	grace := s.cfg.OrphanSweep.Grace
-	if grace <= 0 {
-		grace = defaultOrphanGrace
-	}
+	policy := s.reapPolicy()
+	grace := policy.Grace
 
-	var victims []orphanVictim
+	var nominations []orphanNomination
 
 	s.mu.Lock()
 	if !s.webhookMode {
@@ -2098,8 +2169,8 @@ func (s *Scheduler) sweepOrphanRunners() {
 			continue
 		}
 		if time.Since(rb.dispatchedAt) >= grace {
-			if v, ok := s.reapRunnerLocked(name, rb); ok {
-				victims = append(victims, v)
+			if n, ok := s.nominateRunnerLocked(name, rb); ok {
+				nominations = append(nominations, n)
 			}
 			continue
 		}
@@ -2122,23 +2193,36 @@ func (s *Scheduler) sweepOrphanRunners() {
 			demand[rb.labelSet]--
 			continue
 		}
-		if v, ok := s.reapRunnerLocked(name, rb); ok {
-			v.discharged = true
-			victims = append(victims, v)
+		if n, ok := s.nominateRunnerLocked(name, rb); ok {
+			n.discharged = true
+			nominations = append(nominations, n)
 		}
 	}
 	s.mu.Unlock()
 
+	victims := s.vetoBusyNominations(nominations, policy)
+
 	for _, v := range victims {
-		if v.discharged {
+		switch {
+		case v.escaped:
+			s.cfg.Log.Warn("ESCAPE: destroying a runner the busy check never cleared — it is wedged, or the busy check is broken",
+				"runner", v.name,
+				"dispatched_for_job", v.key.JobID,
+				"busy_verdict", v.verdict,
+				"reason", v.reason,
+				"grace", grace,
+				"hard_bound", policy.HardBound)
+		case v.discharged:
 			s.cfg.Log.Info("retiring discharged runner: its job ran on a same-label sibling and no queued job in that label set needs it",
 				"runner", v.name,
 				"dispatched_for_job", v.key.JobID,
+				"busy_verdict", v.verdict,
 				"detail", "same-label JIT runners are fungible; reconciled on the label set instead of waiting out the grace window")
-		} else {
+		default:
 			s.cfg.Log.Warn("destroying orphaned runner: dispatched but never assigned a job within the grace window",
 				"runner", v.name,
 				"dispatched_for_job", v.key.JobID,
+				"busy_verdict", v.verdict,
 				"grace", grace)
 		}
 		metrics.JobsActive.Dec()