fix(hostutility): cap concurrent agent bootstraps so a cold sweep can finish (#2192)

apps/backend/internal/agent/hostutility/manager.go

@@ -130,6 +130,7 @@ func (m *Manager) Start(ctx context.Context) error {
 	}
 
 	g, gctx := errgroup.WithContext(startCtx)
+	g.SetLimit(maxConcurrentBootstraps)
 	for _, ag := range targets {
 		ag := ag
 		g.Go(func() error {
@@ -520,6 +521,18 @@ func (m *Manager) dropStaleInstance(ctx context.Context, agentType string, inst
 // timeout on the agentctl call.
 const probeTimeout = 60 * time.Second
 
+// maxConcurrentBootstraps caps how many agents are warmed and probed at once at
+// boot. Unbounded, every ACP-capable agent starts together, and each spawns its
+// runtime through `npx -y <package>`, which installs the package when missing.
+// On a cold npm cache they starve each other: measured on Windows with five
+// installed agents, every probe hit the 60s probeTimeout and none succeeded,
+// while the same five capped at two produced two usable results.
+//
+// With a warm cache the cap makes no measurable difference, so this is not a
+// speed-up — it stops a cold sweep from failing wholesale and leaving the
+// capability cache empty for every agent.
+const maxConcurrentBootstraps = 2
+
 // probe runs an ACP probe against the given instance and translates the result
 // into an AgentCapabilities record suitable for the cache.
 func (m *Manager) probe(ctx context.Context, inst *instance, ia agents.InferenceAgent, refresh bool) AgentCapabilities {

apps/backend/internal/agent/hostutility/manager_test.go

@@ -242,6 +242,111 @@ func TestStopCancelsInFlightBootstrap(t *testing.T) {
 	}
 }
 
+func TestStartLimitsConcurrentBootstraps(t *testing.T) {
+	log := newTestLogger(t)
+	reg := registry.NewRegistry(log)
+	started := make(chan string, 8)
+	release := make(chan struct{})
+	var active atomic.Int32
+	var peak atomic.Int32
+	var attempted atomic.Int32
+
+	for i := 0; i < maxConcurrentBootstraps+2; i++ {
+		ag := &trackingInferenceAgent{
+			installedInferenceAgent: installedInferenceAgent{id: "tracking-acp-" + strconv.Itoa(i)},
+			started:                 started,
+			release:                 release,
+			active:                  &active,
+			peak:                    &peak,
+			attempted:               &attempted,
+		}
+		require.NoError(t, reg.Register(ag))
+	}
+
+	mgr := NewManager(reg, "127.0.0.1", 1, nil, log)
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	done := make(chan error, 1)
+	go func() { done <- mgr.Start(ctx) }()
+
+	waitForStartedAgents(t, started, maxConcurrentBootstraps)
+	require.LessOrEqual(t, peak.Load(), int32(maxConcurrentBootstraps))
+	close(release)
+
+	select {
+	case err := <-done:
+		require.NoError(t, err)
+	case <-time.After(time.Second):
+		t.Fatal("host utility bootstrap did not finish after releasing agents")
+	}
+	require.Equal(t, int32(maxConcurrentBootstraps+2), attempted.Load())
+	require.LessOrEqual(t, peak.Load(), int32(maxConcurrentBootstraps))
+	mgr.Stop(context.Background())
+}
+
+func TestStopCancelsQueuedBootstrap(t *testing.T) {
+	log := newTestLogger(t)
+	reg := registry.NewRegistry(log)
+	started := make(chan string, 8)
+	canceled := make(chan string, 8)
+	var active atomic.Int32
+	var peak atomic.Int32
+	var attempted atomic.Int32
+
+	for i := 0; i < maxConcurrentBootstraps+2; i++ {
+		ag := &trackingInferenceAgent{
+			installedInferenceAgent: installedInferenceAgent{id: "queued-acp-" + strconv.Itoa(i)},
+			started:                 started,
+			canceled:                canceled,
+			active:                  &active,
+			peak:                    &peak,
+			attempted:               &attempted,
+		}
+		require.NoError(t, reg.Register(ag))
+	}
+
+	mgr := NewManager(reg, "127.0.0.1", 1, nil, log)
+	ctx, cancel := context.WithCancel(context.Background())
+	done := make(chan error, 1)
+	go func() { done <- mgr.Start(ctx) }()
+
+	waitForStartedAgents(t, started, maxConcurrentBootstraps)
+	mgr.Stop(context.Background())
+
+	select {
+	case err := <-done:
+		require.NoError(t, err)
+	case <-time.After(time.Second):
+		t.Fatal("host utility Start did not return after queued cancellation")
+	}
+	cancel()
+	require.Equal(t, int32(maxConcurrentBootstraps+2), attempted.Load())
+	waitForCanceledAgents(t, canceled, maxConcurrentBootstraps+2)
+	require.LessOrEqual(t, peak.Load(), int32(maxConcurrentBootstraps))
+}
+
+func waitForStartedAgents(t *testing.T, started <-chan string, count int) {
+	t.Helper()
+	for i := 0; i < count; i++ {
+		select {
+		case <-started:
+		case <-time.After(time.Second):
+			t.Fatalf("timed out waiting for bootstrap %d/%d to start", i+1, count)
+		}
+	}
+}
+
+func waitForCanceledAgents(t *testing.T, canceled <-chan string, count int) {
+	t.Helper()
+	for i := 0; i < count; i++ {
+		select {
+		case <-canceled:
+		case <-time.After(time.Second):
+			t.Fatalf("timed out waiting for bootstrap %d/%d to observe cancellation", i+1, count)
+		}
+	}
+}
+
 func TestProbePreservesConfigOptionDescriptions(t *testing.T) {
 	log := newTestLogger(t)
 	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
@@ -631,6 +736,42 @@ type blockingInferenceAgent struct {
 	canceled chan struct{}
 }
 
+type trackingInferenceAgent struct {
+	installedInferenceAgent
+	started   chan<- string
+	canceled  chan<- string
+	release   <-chan struct{}
+	active    *atomic.Int32
+	peak      *atomic.Int32
+	attempted *atomic.Int32
+}
+
+func (a *trackingInferenceAgent) IsInstalled(ctx context.Context) (*agents.DiscoveryResult, error) {
+	a.attempted.Add(1)
+	current := a.active.Add(1)
+	for {
+		previous := a.peak.Load()
+		if current <= previous || a.peak.CompareAndSwap(previous, current) {
+			break
+		}
+	}
+	a.started <- a.id
+	defer a.active.Add(-1)
+
+	if a.release == nil {
+		<-ctx.Done()
+		a.canceled <- a.id
+		return nil, ctx.Err()
+	}
+	select {
+	case <-a.release:
+		return &agents.DiscoveryResult{Available: false}, nil
+	case <-ctx.Done():
+		a.canceled <- a.id
+		return nil, ctx.Err()
+	}
+}
+
 func (a *blockingInferenceAgent) ID() string          { return "blocking-acp" }
 func (a *blockingInferenceAgent) Name() string        { return "Blocking ACP" }
 func (a *blockingInferenceAgent) DisplayName() string { return "Blocking ACP" }