From bececdb5f704e8bc7404dcf777cc0333f8605bd5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jo=C3=A3o=20Manso?= Date: Mon, 3 Aug 2026 12:16:39 +0100 Subject: [PATCH] fix(hostutility): cap concurrent agent bootstraps so a cold sweep can finish (#2192) --- .../internal/agent/hostutility/manager.go | 13 ++ .../agent/hostutility/manager_test.go | 141 ++++++++++++++++++ 2 files changed, 154 insertions(+) diff --git a/apps/backend/internal/agent/hostutility/manager.go b/apps/backend/internal/agent/hostutility/manager.go index 317d8e0560..92d7369d2b 100644 --- a/apps/backend/internal/agent/hostutility/manager.go +++ b/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 `, 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 { diff --git a/apps/backend/internal/agent/hostutility/manager_test.go b/apps/backend/internal/agent/hostutility/manager_test.go index 733bbefe36..8babd11bb8 100644 --- a/apps/backend/internal/agent/hostutility/manager_test.go +++ b/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" }