Avoid empty managed Runtime scan intervals

CONTRIBUTING.md

@@ -303,8 +303,10 @@ The allocation lifecycle owns pending/running/complete initialization. Authentic
 may connect the daemon during initialization; execution bindings, native preparation,
 live Files and connected publication wait for completion. Keep Provider bootstrap
 settlement distinct. Advance at most one bounded initialization operation per full
-maintenance scan,
-using process-local progress and the existing lifecycle gate. A recovered or uncertain
+maintenance scan. At allocation EOF, begin the next page in the same call rather
+than consume an observation interval on an empty page. Refill at most once, retain
+the 32-allocation per-call bound and the five-second ticker, and never loop on an
+empty store. Use process-local progress and the existing lifecycle gate. A recovered or uncertain
 running installation fails and uses existing cleanup, without replaying writes.
 Completed environments never reinstall initial files on reconnect or native recovery.
 Provider RunCommand carries bounded stdin, not confidential argv. Only fixed trusted

services/agents-api/internal/execution/runtime_lifecycle.go

@@ -188,14 +188,22 @@ func (w *Worker) ReconcileManagedRuntimes(ctx context.Context) error {
 		return err
 	}
 	if len(rows) == 0 {
+		wrapped := r.cursor != ""
 		r.cursor = ""
 		if err := r.advanceInitialization(ctx); err != nil {
 			if ownership := r.store.CheckExecutionOwnership(ctx); ownership != nil {
 				return ownership
 			}
 			log.Ctx(ctx).Warn("managed Runtime file initialization incomplete")
 		}
-		return r.provisionPending(ctx)
+		if wrapped {
+			// Service the next page now instead of spending a ticker interval on EOF.
+			// Refill only once so an empty store still returns without spinning.
+			rows, err = r.store.ListRuntimeAllocations(ctx, "")
+			if err != nil {
+				return err
+			}
+		}
 	}
 	for _, owner := range rows {
 		r.cursor = owner.ID

services/agents-api/internal/store/runtime_initialization_test.go

@@ -92,18 +92,25 @@ func TestManagedInitialFilesGateFairnessCompletionAndRestart(t *testing.T) {
 			if err != nil || !ok || credential.ID != owner.DeviceID {
 				t.Fatal("pending initialization blocks daemon authentication")
 			}
+			lastStepGets := 0
 			p.check = func() {
+				if p.gets-lastStepGets < 33 {
+					t.Fatal("initialization advanced before a full allocation scan", p.gets-lastStepGets)
+				}
+				lastStepGets = p.gets
 				if _, err := s.GetSessionDevice(t.Context(), tenant, session.ID); !errors.Is(err, store.ErrNotFound) {
 					t.Fatal("pending file access", err)
 				}
 				if _, err := s.GetSessionExecutionBinding(t.Context(), tenant, session.ID); !errors.Is(err, store.ErrNotFound) {
 					t.Fatal("premature native preparation", err)
 				}
 			}
-			// Another allocation is observed between file steps, rather than after the full batch.
-			otherTenant, _, otherEnv := managedSession(t, s)
-			if _, err := w.ProvisionEnvironment(t.Context(), otherTenant, otherEnv.ID, key); err != nil {
-				t.Fatal(err)
+			// A full page of other allocations is serviced between initialization steps.
+			for range 32 {
+				otherTenant, _, otherEnv := managedSession(t, s)
+				if _, err := w.ProvisionEnvironment(t.Context(), otherTenant, otherEnv.ID, key); err != nil {
+					t.Fatal(err)
+				}
 			}
 			for n := 0; p.writes == 0 && n < 100; n++ {
 				if err := w.ReconcileManagedRuntimes(t.Context()); err != nil {

services/agents-api/internal/store/runtime_scan_test.go

@@ -0,0 +1,123 @@
+package store_test
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"slices"
+	"testing"
+	"time"
+
+	"github.com/google/uuid"
+
+	"github.com/MiniMax-AI-Dev/parsar/services/agents-api/internal/sandbox"
+	"github.com/MiniMax-AI-Dev/parsar/services/agents-api/internal/store"
+)
+
+type scanProvider struct {
+	lifecycleProvider
+	observed []string
+}
+
+func (p *scanProvider) GetInfo(ctx context.Context, ref sandbox.Reference) (sandbox.Info, error) {
+	p.observed = append(p.observed, ref.AllocationID)
+	return p.lifecycleProvider.GetInfo(ctx, ref)
+}
+
+func TestManagedRuntimeScanWrapServicesNextPage(t *testing.T) {
+	for _, count := range []int{0, 1, 31, 32, 33, 65} {
+		t.Run(fmt.Sprint(count), func(t *testing.T) {
+			s, _ := store.NewManagedTestStore(t)
+			p := &scanProvider{lifecycleProvider: lifecycleProvider{resources: map[string]sandbox.Info{}}}
+			key := uuid.NewString()
+			w, _ := managedWorker(t, s, key, p)
+			var ids []string
+			for range count {
+				tenant, _, env := managedSession(t, s)
+				owner, err := w.ProvisionEnvironment(t.Context(), tenant, env.ID, key)
+				if err != nil {
+					t.Fatal(err)
+				}
+				ids = append(ids, owner.ID)
+			}
+			slices.Sort(ids)
+			// Each call must service the next bounded page, including the call
+			// that wraps at EOF. Empty stores must return without spinning.
+			pages := max(1, (count+31)/32)
+			for call := 0; call < pages*3; call++ {
+				p.observed = nil
+				ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second)
+				err := w.ReconcileManagedRuntimes(ctx)
+				cancel()
+				if err != nil {
+					t.Fatal(err)
+				}
+				start := (call % pages) * 32
+				want := ids[start:min(start+32, count)]
+				if !slices.Equal(p.observed, want) {
+					t.Fatalf("call %d observed %v, want %v", call, p.observed, want)
+				}
+			}
+			if p.creates != count || p.kills != 0 {
+				t.Fatal("scanning changed resource ownership", p.creates, p.kills)
+			}
+		})
+	}
+}
+
+func TestManagedRuntimeScanEmptyAfterCleanupAndCanceledCall(t *testing.T) {
+	s, _ := store.NewManagedTestStore(t)
+	p := &scanProvider{lifecycleProvider: lifecycleProvider{resources: map[string]sandbox.Info{}}}
+	key := uuid.NewString()
+	w, _ := managedWorker(t, s, key, p)
+	tenant, session, env := managedSession(t, s)
+	owner, err := w.ProvisionEnvironment(t.Context(), tenant, env.ID, key)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := w.ReconcileManagedRuntimes(t.Context()); err != nil {
+		t.Fatal(err)
+	}
+	p.observed = nil
+	ctx, cancel := context.WithCancel(t.Context())
+	cancel()
+	if err := w.ReconcileManagedRuntimes(ctx); !errors.Is(err, context.Canceled) {
+		t.Fatal("canceled scan", err)
+	}
+	if len(p.observed) != 0 {
+		t.Fatal("canceled scan reached provider")
+	}
+	if err := s.DeleteSession(t.Context(), tenant, session.ID); err != nil {
+		t.Fatal(err)
+	}
+	// EOF must not delay cleanup to a second call, and cancellation must
+	// release the lifecycle gate.
+	if err := w.ReconcileManagedRuntimes(t.Context()); err != nil {
+		t.Fatal(err)
+	}
+	got, err := s.GetRuntimeAllocation(t.Context(), tenant, env.ID)
+	if err != nil || got.State != "released" || p.kills != 1 {
+		t.Fatal("cleanup delayed at EOF", got, err, p.kills)
+	}
+	for range 2 {
+		p.observed = nil
+		ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second)
+		err := w.ReconcileManagedRuntimes(ctx)
+		cancel()
+		if err != nil || len(p.observed) != 0 {
+			t.Fatal("released allocation reobserved", owner.ID, err, p.observed)
+		}
+	}
+	// A new allocation remains discoverable after the store becomes empty.
+	tenant, _, env = managedSession(t, s)
+	next, err := w.ProvisionEnvironment(t.Context(), tenant, env.ID, key)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := w.ReconcileManagedRuntimes(t.Context()); err != nil {
+		t.Fatal(err)
+	}
+	if !slices.Equal(p.observed, []string{next.ID}) {
+		t.Fatal("new allocation stranded after empty scan", p.observed)
+	}
+}