Do not place on a fleet that has not reported in yet

A control plane that has just started knows nothing, and its first
reconcile tick runs before any agent has registered. "needs 2 nodes on one
fabric, but no node reports a fabric" in the log is that, and it looked
harmless because adoption catches up.

What is not harmless is placing on the fraction of the fleet that happened
to report first. After one restart both replicas of qwen2.5-0.5b landed on
node04 — the only node registered at that instant — when four were coming,
and nothing ever moved them, because the desired count was met.

Placement now waits, and leaves by either of two doors, because either
alone is wrong. Waiting for the window to expire would cost that much
downtime on every restart even when the whole fleet reconnected in a
second; waiting only for the node count to stop changing would hang forever
on a cluster whose nodes never all come back. So: the count holds steady for
a quarter of the window, or the window expires.

Only placement waits. Node expiry, stuck-replica expiry, draining and
reaping all stay correct on a partial fleet, and adoption in particular must
not wait — it is how a restart recovers what is already running, and
delaying it would leave live engines unaccounted for exactly while the
reconciler decides what to build.

The window is paid once per process. A node joining a running cluster is an
ordinary event, not a reason to stop placing.

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

internal/control/reconciler/reconciler.go

@@ -66,6 +66,22 @@ type Config struct {
 	PortRange [2]int
 	// MgmtInterface is the ethernet interface used for NCCL bootstrap.
 	MgmtInterface string
+	// SettleFor is how long after start to leave placement alone while agents
+	// reconnect.
+	//
+	// A control plane that has just started knows nothing, and the first
+	// reconcile tick runs before any agent has registered. It sees an empty
+	// fleet and acts on it: "needs 2 nodes on one fabric, but no node reports a
+	// fabric" is that, and it is harmless only because adoption catches up.
+	// What is not harmless is placing on the fraction of the fleet that
+	// happens to have reported first — after one restart both replicas of a
+	// model landed on the single node that had registered, when four were
+	// coming.
+	//
+	// Adoption still runs throughout; only placement waits. The window ends
+	// early once the fleet stops growing, so a small cluster does not sit idle
+	// for the whole duration.
+	SettleFor time.Duration
 }
 
 // DefaultConfig returns sensible defaults.
@@ -76,6 +92,7 @@ func DefaultConfig() Config {
 		MaxBackoff:       10 * time.Minute,
 		UnreachableAfter: 15 * time.Second,
 		LostAfter:        60 * time.Second,
+		SettleFor:        20 * time.Second,
 		DrainGrace:       5 * time.Minute,
 		PortRange:        [2]int{20000, 20999},
 		MgmtInterface:    "eth0",
@@ -93,6 +110,11 @@ func (c *Config) normalize() {
 	if c.MaxBackoff < c.BaseBackoff {
 		c.MaxBackoff = max(c.BaseBackoff, d.MaxBackoff)
 	}
+	if c.SettleFor < 0 {
+		c.SettleFor = 0
+	} else if c.SettleFor == 0 {
+		c.SettleFor = d.SettleFor
+	}
 	if c.UnreachableAfter <= 0 {
 		c.UnreachableAfter = d.UnreachableAfter
 	}
@@ -138,6 +160,12 @@ type Reconciler struct {
 	stop chan struct{}
 	done chan struct{}
 	once sync.Once
+
+	// startedAt and lastNodeCount track the settling window: placement waits
+	// until the fleet has stopped growing, or the window expires.
+	startedAt     time.Time
+	lastNodeCount int
+	steadySince   time.Time
 }
 
 // New builds a reconciler.
@@ -158,11 +186,15 @@ func New(cfg Config, cat *catalog.Catalog, inv *inventory.Store, disp Dispatcher
 		wake:          make(chan struct{}, 1),
 		stop:          make(chan struct{}),
 		done:          make(chan struct{}),
+		startedAt:     time.Now(),
 	}
 }
 
 // SetClock replaces the time source, for tests.
-func (r *Reconciler) SetClock(now func() time.Time) { r.now = now }
+func (r *Reconciler) SetClock(now func() time.Time) {
+	r.now = now
+	r.startedAt = now()
+}
 
 // Start runs the loop until Close.
 func (r *Reconciler) Start(ctx context.Context) {
@@ -212,11 +244,51 @@ func (r *Reconciler) Reconcile(ctx context.Context) {
 	r.drainCordonedNodes(ctx)
 	r.reapFailed(ctx)
 
+	// Everything above is about what is already running and stays correct on a
+	// partial fleet. Placement is the part that must not act on one.
+	if !r.fleetHasSettled() {
+		return
+	}
+
 	for _, m := range r.cat.Models() {
 		r.reconcileModel(ctx, m)
 	}
 }
 
+// fleetHasSettled reports whether placement may run yet.
+//
+// Two ways out, because either alone is wrong. Waiting for the window to expire
+// would make every restart cost that much downtime even when the whole fleet
+// reconnected in a second. Waiting only for the count to stop changing would
+// hang forever on a cluster whose nodes never all come back.
+func (r *Reconciler) fleetHasSettled() bool {
+	if r.cfg.SettleFor <= 0 {
+		return true
+	}
+	now := r.now()
+	if now.Sub(r.startedAt) >= r.cfg.SettleFor {
+		return true
+	}
+
+	n := len(r.inv.Nodes())
+	if n != r.lastNodeCount {
+		r.lastNodeCount, r.steadySince = n, now
+		return false
+	}
+	// Steady for a moment, and at least one node present. A count that has been
+	// zero since start is not a settled fleet, it is one that has not arrived.
+	if n == 0 {
+		return false
+	}
+	settled := now.Sub(r.steadySince) >= r.cfg.SettleFor/4
+	if settled {
+		r.log.Info("fleet settled; placement resuming",
+			"nodes", n, "waited", now.Sub(r.startedAt).Round(time.Millisecond))
+		r.startedAt = time.Time{} // never wait again in this process
+	}
+	return settled
+}
+
 // expireStuckReplicas fails replicas that never finished coming up.
 //
 // Every other path out of Starting depends on the agent reporting something. A

internal/control/reconciler/reconciler_test.go

@@ -141,6 +141,14 @@ func newHarness(t *testing.T, m catalog.Model, nodes ...inventory.Node) *harness
 	for _, n := range nodes {
 		touchAt(inv, n.ID, c.now())
 	}
+	// Past the startup settling window. These nodes are all here — the harness
+	// built the whole fleet before anything ran — which is the state settling
+	// waits for, not one it should be made to simulate in every test. The
+	// window's own behaviour is tested separately.
+	c.advance(cfg.SettleFor)
+	for _, n := range nodes {
+		touchAt(inv, n.ID, c.now())
+	}
 	return &harness{r: r, inv: inv, disp: disp, clock: c}
 }
 
@@ -992,3 +1000,117 @@ func TestAHalfThatNeverArrivesStillFails(t *testing.T) {
 		t.Errorf("node has %d free GPUs; the abandoned half still holds them", free)
 	}
 }
+
+// --- the startup settling window ---
+
+func settlingHarness(t *testing.T, m catalog.Model) (*Reconciler, *inventory.Store, *fakeDispatcher, *clock) {
+	t.Helper()
+	cat, err := catalog.New([]catalog.Model{m})
+	if err != nil {
+		t.Fatal(err)
+	}
+	inv := inventory.NewStore()
+	disp := &fakeDispatcher{}
+	c := newClock()
+	r := New(DefaultConfig(), cat, inv, disp, slog.New(slog.DiscardHandler))
+	r.SetClock(c.now)
+	return r, inv, disp, c
+}
+
+func addNode(inv *inventory.Store, c *clock, n inventory.Node) {
+	inv.Upsert(n)
+	touchAt(inv, n.ID, c.now())
+}
+
+func TestPlacementWaitsForTheFleetToArrive(t *testing.T) {
+	// A control plane that has just started knows nothing, and the first tick
+	// runs before any agent has registered. Placing then puts everything on
+	// whichever fraction reported first: after one restart both replicas of a
+	// model landed on the single node that had registered, when four were
+	// coming.
+	r, inv, disp, c := settlingHarness(t, testModel(2))
+
+	// Nothing registered yet.
+	r.Reconcile(t.Context())
+	if got := len(disp.startedShards()); got != 0 {
+		t.Fatalf("placed %d shards on an empty fleet", got)
+	}
+
+	// One node of four.
+	addNode(inv, c, gpuNode("1", 8))
+	r.Reconcile(t.Context())
+	if got := len(disp.startedShards()); got != 0 {
+		t.Fatalf("placed %d shards while the fleet was still arriving", got)
+	}
+
+	// The rest arrive.
+	for _, id := range []string{"2", "3", "4"} {
+		addNode(inv, c, gpuNode(id, 8))
+	}
+	r.Reconcile(t.Context())
+	if got := len(disp.startedShards()); got != 0 {
+		t.Fatalf("placed %d shards the instant the last node appeared", got)
+	}
+
+	// Steady for a moment: now it may act, and on the whole fleet.
+	c.advance(DefaultConfig().SettleFor / 4)
+	for _, id := range []string{"1", "2", "3", "4"} {
+		touchAt(inv, id, c.now())
+	}
+	r.Reconcile(t.Context())
+	if got := len(disp.startedShards()); got == 0 {
+		t.Fatal("placement never resumed after the fleet settled")
+	}
+}
+
+func TestPlacementDoesNotWaitForeverForNodesThatNeverCome(t *testing.T) {
+	// Waiting only for the count to stop changing would hang forever on a
+	// cluster whose nodes never all come back.
+	r, inv, disp, c := settlingHarness(t, testModel(1))
+	addNode(inv, c, gpuNode("1", 8))
+
+	c.advance(DefaultConfig().SettleFor)
+	touchAt(inv, "1", c.now())
+	r.Reconcile(t.Context())
+
+	if got := len(disp.startedShards()); got == 0 {
+		t.Error("placement never resumed once the window expired")
+	}
+}
+
+func TestSettlingIsOnlyEverPaidOnce(t *testing.T) {
+	// A node joining later in the day is an ordinary event, not a reason to
+	// stop placing while the cluster is running.
+	r, inv, disp, c := settlingHarness(t, testModel(1))
+	addNode(inv, c, gpuNode("1", 8))
+	c.advance(DefaultConfig().SettleFor)
+	touchAt(inv, "1", c.now())
+	r.Reconcile(t.Context())
+	if len(disp.startedShards()) == 0 {
+		t.Fatal("nothing placed after the window")
+	}
+
+	disp.reset()
+	addNode(inv, c, gpuNode("2", 8))
+	if !r.fleetHasSettled() {
+		t.Error("a node joining a running cluster restarted the settling window")
+	}
+}
+
+func TestAdoptionIsNotBlockedByTheSettlingWindow(t *testing.T) {
+	// Only placement waits. Adoption is how a restart recovers what is already
+	// running, and delaying it would leave live engines unaccounted for exactly
+	// when the reconciler is deciding what to build.
+	r, inv, _, c := settlingHarness(t, testModel(1))
+	addNode(inv, c, gpuNode("1", 8))
+
+	unclaimed := r.Adopt("1", []AdoptedShard{
+		adopted("llama-99", "llama-99-r0", "1", 0, []int{0, 1, 2, 3}),
+	})
+	if len(unclaimed) != 0 {
+		t.Fatalf("adoption was refused during settling: %+v", unclaimed)
+	}
+	if got := len(r.Replicas()); got != 1 {
+		t.Fatalf("expected the replica to be adopted, got %d", got)
+	}
+}