Treat a zombie as dead, and test fairness where it is deterministic

CI went red on Linux with two failures that never appear on macOS. Both were
worth the trip.

The first is a real bug. processAlive used kill(pid, 0), which succeeds on a
zombie — a process that has exited and is only waiting to be reaped. That
happens for real whenever an engine's parent dies before it does and nothing
reaps the orphan, which in a container is the normal case: PID 1 is usually the
workload rather than an init. The supervisor would then report a dead adopted
engine as running forever, and hold its GPUs against every future placement.
Zombies are now detected through procfs on Linux, where the answer exists, and
the check degrades to its old behaviour elsewhere rather than to something worse.

The second was a bad test. TestTenantFairnessWithinBand asserted on the order
goroutines sent to a channel after being admitted, which is the order the Go
scheduler woke them, not the order the dispatcher chose. Under load it measured
the runtime. The rotation itself lives in bandQueue.pop, which is pure, so it is
tested there — including what happens when a tenant runs out mid-rotation and
when a cancelled waiter is removed. The controller-level test now asserts the
thing that is actually stable: nobody is starved.

Verified with three consecutive `go test -race ./...` runs in a Linux container.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

internal/agent/supervisor/process_unix.go

@@ -1,7 +1,9 @@
 package supervisor
 
 import (
+	"bytes"
 	"os"
+	"strconv"
 	"syscall"
 )
 
@@ -22,13 +24,42 @@ func killGroup(pid int, sig syscall.Signal) error {
 	return nil
 }
 
-// processAlive reports whether a pid exists.
+// processAlive reports whether a pid belongs to a process that is still doing
+// something.
+//
+// A zombie is not. It has exited and is only waiting to be reaped, but
+// kill(pid, 0) still succeeds on it — so a naive check reports a dead engine as
+// running forever. That happens for real whenever an engine's parent dies
+// before it does and nothing reaps the orphan: in a container, PID 1 is usually
+// the workload rather than an init, and orphans simply accumulate.
 func processAlive(pid int) bool {
 	if pid <= 0 {
 		return false
 	}
 	// Signal 0 performs the permission and existence checks without delivering
 	// anything.
-	err := syscall.Kill(pid, 0)
-	return err == nil || err == syscall.EPERM
+	if err := syscall.Kill(pid, 0); err != nil {
+		// EPERM means it exists but belongs to someone else, which for our
+		// purposes is still "there".
+		return err == syscall.EPERM
+	}
+	return !isZombie(pid)
+}
+
+// isZombie reports whether a pid has exited but not been reaped.
+//
+// Only Linux exposes this; elsewhere there is no procfs and the answer is
+// false, which degrades to the old behaviour rather than to something worse.
+func isZombie(pid int) bool {
+	b, err := os.ReadFile("/proc/" + strconv.Itoa(pid) + "/stat")
+	if err != nil {
+		return false
+	}
+	// The comm field is parenthesised and may itself contain spaces or
+	// parentheses, so the state character is found from the last ')'.
+	i := bytes.LastIndexByte(b, ')')
+	if i < 0 || i+2 >= len(b) {
+		return false
+	}
+	return b[i+2] == 'Z'
 }

internal/router/admission/admission_test.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"strings"
 	"sync"
 	"sync/atomic"
 	"testing"
@@ -213,71 +214,136 @@ func TestClientCancellationLeavesQueue(t *testing.T) {
 	t.Fatal("a cancelled request stayed in the queue, occupying capacity")
 }
 
-func TestTenantFairnessWithinBand(t *testing.T) {
-	// One tenant flooding the band must not take every dispatch turn.
+// TestBandQueueRotatesTenants is where the fairness property actually lives.
+//
+// It is tested on the queue rather than through the controller on purpose: at
+// the controller level the observable order is the order goroutines wake up,
+// which the scheduler decides, not the order the dispatcher admitted them. A
+// test asserting on that is measuring the Go runtime.
+func TestBandQueueRotatesTenants(t *testing.T) {
+	q := newBandQueue(Band{Name: "bronze"})
+
+	// One tenant floods, then a quiet one arrives strictly behind it.
+	for range 10 {
+		q.push(&waiter{tenant: "noisy", model: "m", ready: make(chan struct{})})
+	}
+	q.push(&waiter{tenant: "quiet", model: "m", ready: make(chan struct{})})
+
+	var order []string
+	for q.len() > 0 {
+		order = append(order, q.pop().tenant)
+	}
+
+	if len(order) != 11 {
+		t.Fatalf("popped %d waiters, pushed 11", len(order))
+	}
+	// FIFO would put the quiet tenant last. Round-robin gives it the second
+	// turn: one noisy, one quiet, then the rest of the flood.
+	if order[0] != "noisy" || order[1] != "quiet" {
+		t.Errorf("rotation order starts %v; a flood is taking consecutive turns", order[:3])
+	}
+	quiet := 0
+	for _, v := range order {
+		if v == "quiet" {
+			quiet++
+		}
+	}
+	if quiet != 1 {
+		t.Errorf("the quiet tenant was admitted %d times", quiet)
+	}
+}
+
+func TestBandQueueRotationIsStable(t *testing.T) {
+	// Three tenants with unequal backlogs must still take turns, and a tenant
+	// that runs out must drop cleanly out of the rotation.
+	q := newBandQueue(Band{Name: "bronze"})
+	for _, tenant := range []string{"a", "a", "a", "b", "c"} {
+		q.push(&waiter{tenant: tenant, model: "m", ready: make(chan struct{})})
+	}
+
+	var order []string
+	for q.len() > 0 {
+		order = append(order, q.pop().tenant)
+	}
+	if strings.Join(order, "") != "abcaa" {
+		t.Errorf("order=%v, expected a,b,c then a's remaining two", order)
+	}
+}
+
+func TestBandQueueRemoveKeepsRotation(t *testing.T) {
+	// A cancelled request leaves the queue; the rotation must not skip or
+	// repeat a tenant because of it.
+	q := newBandQueue(Band{Name: "bronze"})
+	a1 := &waiter{tenant: "a", model: "m", ready: make(chan struct{})}
+	b1 := &waiter{tenant: "b", model: "m", ready: make(chan struct{})}
+	a2 := &waiter{tenant: "a", model: "m", ready: make(chan struct{})}
+	q.push(a1)
+	q.push(b1)
+	q.push(a2)
+
+	if !q.remove(b1) {
+		t.Fatal("remove reported the waiter was not queued")
+	}
+	if q.len() != 2 {
+		t.Fatalf("len=%d after removing one of three", q.len())
+	}
+
+	var order []string
+	for q.len() > 0 {
+		order = append(order, q.pop().tenant)
+	}
+	if strings.Join(order, "") != "aa" {
+		t.Errorf("order=%v after the only b was removed", order)
+	}
+}
+
+// TestNoTenantIsStarved is the controller-level companion: it asserts that
+// everything queued is eventually admitted, without asserting an order the Go
+// scheduler controls.
+func TestNoTenantIsStarved(t *testing.T) {
 	s := &fixedSat{}
 	s.set(1.0) // hold everything initially
 	cfg := testConfig()
-	cfg.QueueTTL = 5 * time.Second
+	cfg.QueueTTL = 10 * time.Second
 	c := newController(t, cfg, s.fn)
 
 	const flood = 40
+	var noisy, quiet atomic.Int64
 	var wg sync.WaitGroup
-	order := make(chan string, flood+4)
 
-	for i := range flood {
-		wg.Add(1)
-		go func() {
-			defer wg.Done()
-			if _, err := c.Acquire(t.Context(), Request{Model: "m", Band: "bronze", Tenant: "noisy"}); err == nil {
-				order <- "noisy"
-			}
-			_ = i
-		}()
+	admit := func(tenant string, counter *atomic.Int64) {
+		defer wg.Done()
+		if _, err := c.Acquire(t.Context(), Request{Model: "m", Band: "bronze", Tenant: tenant}); err == nil {
+			counter.Add(1)
+		}
 	}
-	// Let the flood queue up first, so the quiet tenant is strictly behind it.
-	deadline := time.Now().Add(2 * time.Second)
+
+	for range flood {
+		wg.Add(1)
+		go admit("noisy", &noisy)
+	}
+	deadline := time.Now().Add(5 * time.Second)
 	for time.Now().Before(deadline) && c.Depths()[3].Depth < flood {
 		time.Sleep(2 * time.Millisecond)
 	}
 
 	for range 4 {
 		wg.Add(1)
-		go func() {
-			defer wg.Done()
-			if _, err := c.Acquire(t.Context(), Request{Model: "m", Band: "bronze", Tenant: "quiet"}); err == nil {
-				order <- "quiet"
-			}
-		}()
+		go admit("quiet", &quiet)
+	}
+	deadline = time.Now().Add(5 * time.Second)
+	for time.Now().Before(deadline) && c.Depths()[3].Depth < flood+4 {
+		time.Sleep(2 * time.Millisecond)
 	}
-	time.Sleep(30 * time.Millisecond)
 
 	s.set(0.0) // open the gate
 	wg.Wait()
-	close(order)
 
-	var seen []string
-	for v := range order {
-		seen = append(seen, v)
+	if noisy.Load() != flood {
+		t.Errorf("the flooding tenant got %d of %d admissions", noisy.Load(), flood)
 	}
-	if len(seen) != flood+4 {
-		t.Fatalf("expected %d admissions, got %d", flood+4, len(seen))
-	}
-
-	// With FIFO the quiet tenant would be admitted 41st at best. Round-robin
-	// should interleave it into the first handful.
-	firstQuiet := -1
-	for i, v := range seen {
-		if v == "quiet" {
-			firstQuiet = i
-			break
-		}
-	}
-	if firstQuiet < 0 {
-		t.Fatal("the quiet tenant was never admitted")
-	}
-	if firstQuiet > 8 {
-		t.Errorf("quiet tenant admitted at position %d; a flood is monopolising the band", firstQuiet)
+	if quiet.Load() != 4 {
+		t.Errorf("the quiet tenant got %d of 4 admissions; it was starved", quiet.Load())
 	}
 }