feat(scheduler): add resource-bounded grouped round-robin scheduling

services/README.md

@@ -10,7 +10,7 @@ Go implementation of a distributed Gateway and pluggable Scheduler for AgentENV.
 - Gateway resolves `GET /nodes/{id}` via scheduler and proxies to the target node.
 - Gateway routes sandbox requests by existing sandbox-to-node binding.
 - Scheduler exposes gRPC API and supports pluggable strategy providers.
-- Built-in strategies in v1: round_robin and random.
+- Built-in strategies in v1: round_robin, random, and grouped_round_robin.
 - Scheduler supports both static node configuration and Kubernetes EndpointSlice discovery.
 - Scheduler sandbox binding store can be in-memory or Redis-backed.
 - Scheduler can run as a primary read/write service or as query-only replicas that serve only `LookupNode` from Redis.
@@ -98,6 +98,9 @@ General config notes:
 - `SCHEDULER_REDIS_ADDR=<addr>` overrides `scheduler.redis_addr` from the environment.
 - `SCHEDULER_ARTIFACT_STORE_CAPACITY=<count>` overrides `scheduler.artifact_store_capacity` from the environment.
 - `SCHEDULER_ARTIFACT_LOOKUP_NODE_LIMIT=<count>` overrides `scheduler.artifact_lookup_node_limit` from the environment.
+- `SCHEDULER_GROUPED_ROUND_ROBIN_MAX_SANDBOX_COUNT=<count>` overrides `scheduler.grouped_round_robin.max_sandbox_count`.
+- `SCHEDULER_GROUPED_ROUND_ROBIN_MAX_CPU_COUNT=<count>` overrides `scheduler.grouped_round_robin.max_cpu_count`.
+- `SCHEDULER_GROUPED_ROUND_ROBIN_MAX_MEMORY_MB=<count>` overrides `scheduler.grouped_round_robin.max_memory_mb`.
 
 ### Scheduling strategy
 
@@ -107,8 +110,28 @@ General config notes:
 |---|---|
 | `round_robin` (default) | Cycles through eligible nodes in stable order |
 | `random` | Picks a uniformly random eligible node |
+| `grouped_round_robin` | Keeps requests for the same image/template in a resource-bounded group, then assigns the next group by global round-robin |
 
-The strategy interface receives `RichNode` values that carry the node identity (ID + endpoint) together with the latest heartbeat `NodeSnapshot` (sandbox counts, CPU, memory, disk metrics). Current built-in strategies ignore the snapshot, but custom strategy implementations can use it for load-aware decisions.
+The strategy interface receives `RichNode` values that carry the node identity (ID + endpoint) together with the latest heartbeat `NodeSnapshot` (sandbox counts, CPU, memory, disk metrics).
+
+The `grouped_round_robin` strategy uses the rootfs image reference from `POST /sandboxes-cold` or the exact template reference from `POST /sandboxes` as its grouping key. Requests for the same key stay on the current group's node until adding another request would exceed any configured group limit. A full group is closed permanently, and the next group is assigned to the next eligible node through one global round-robin cursor shared by all keys. If the open group's node is no longer ready or was removed by the node resource filter, the group closes early.
+
+Group accounting occurs atomically during scheduling, before the runtime finishes creating the sandbox, so bursts cannot overfill an open group while heartbeat metrics lag. Groups are advisory and in-memory; scheduler restart forgets them. The strategy requires a fresh READY heartbeat and skips nodes with missing or stale telemetry.
+
+`scheduler.grouped_round_robin.max_sandbox_count` is required and must be greater than zero when `scheduler.strategy` is `grouped_round_robin`. CPU and memory limits are optional additional bounds for cold-start requests:
+
+```json
+"strategy": "grouped_round_robin",
+"grouped_round_robin": {
+  "max_sandbox_count": 4,
+  "max_cpu_count": 8,
+  "max_memory_mb": 16384
+}
+```
+
+When a cold-start request omits `cpuCount` or `memoryMB` while the corresponding group limit is enabled, the scheduler charges that dimension at the full configured group limit. Runtime defaults are node-local and are not available in the gateway hint, so this conservative behavior closes that request's group instead of allowing an unknown default to exceed the budget.
+
+Template-based create requests do not carry CPU or memory values, so their groups are bounded by `max_sandbox_count` only. Image tags and template aliases are used exactly as supplied and may be mutable; a stale grouping hint can reduce cache affinity, but cannot bypass group or node resource limits.
 
 ### Node resource limit
 

services/api/proto/scheduler.proto

@@ -51,6 +51,9 @@ message NewColdSandboxHint {
 message NewSandboxHint {
   // Sandbox metadata key/value pairs parsed from the request body.
   map<string, string> metadata = 1;
+  // Template or snapshot reference supplied by POST /sandboxes. The grouped
+  // round-robin strategy treats the exact reference as a workload identity.
+  string template_id = 2;
 }
 
 message ScheduleRequest {

services/gateway/internal/schedule_hint.go

@@ -134,7 +134,8 @@ func parseNewColdSandboxHint(body []byte) *schedulerv1.NewColdSandboxHint {
 // newSandboxBody mirrors the subset of NewSandbox (src/api/openapi.yml) that is
 // relevant for scheduling.
 type newSandboxBody struct {
-	Metadata map[string]string `json:"metadata"`
+	TemplateID string            `json:"templateID"`
+	Metadata   map[string]string `json:"metadata"`
 }
 
 // parseNewSandboxHint extracts the structured sandbox hint from the request
@@ -149,6 +150,7 @@ func parseNewSandboxHint(body []byte) *schedulerv1.NewSandboxHint {
 	if err := json.Unmarshal(body, &parsed); err != nil {
 		return hint
 	}
+	hint.TemplateId = parsed.TemplateID
 	hint.Metadata = parsed.Metadata
 	return hint
 }

services/gateway/internal/schedule_hint_test.go

@@ -35,6 +35,9 @@ func TestBuildScheduleHintNewSandbox(t *testing.T) {
 	if hint.GetNewColdSandbox() != nil {
 		t.Fatalf("did not expect cold sandbox hint")
 	}
+	if got := hint.GetNewSandbox().GetTemplateId(); got != "tmpl" {
+		t.Fatalf("template_id = %q, want tmpl", got)
+	}
 
 	// Body must remain available for the upstream request.
 	body, err := io.ReadAll(r.Body)
@@ -158,6 +161,25 @@ func TestParseNewColdSandboxHint(t *testing.T) {
 	})
 }
 
+func TestParseNewSandboxHint(t *testing.T) {
+	t.Run("template and metadata", func(t *testing.T) {
+		hint := parseNewSandboxHint([]byte(`{"templateID":"tmpl","metadata":{"team":"infra"}}`))
+		if got := hint.GetTemplateId(); got != "tmpl" {
+			t.Fatalf("template_id = %q, want tmpl", got)
+		}
+		if got := hint.GetMetadata()["team"]; got != "infra" {
+			t.Fatalf("metadata team = %q, want infra", got)
+		}
+	})
+
+	t.Run("malformed json", func(t *testing.T) {
+		hint := parseNewSandboxHint([]byte("{not json"))
+		if hint.GetTemplateId() != "" || len(hint.GetMetadata()) != 0 {
+			t.Fatalf("expected empty best-effort hint, got %v", hint)
+		}
+	})
+}
+
 func TestCaptureRequestBodyNil(t *testing.T) {
 	r := newHintRequest(t, http.MethodPost, "/sandboxes-cold", "")
 	body, err := captureRequestBody(r)

services/scheduler/cmd/main.go

@@ -69,7 +69,14 @@ func main() {
 		svc := scheduler.NewService(
 			logger,
 			registry,
-			scheduler.NewStrategy(cfg.Scheduler.Strategy),
+			scheduler.NewStrategy(
+				cfg.Scheduler.Strategy,
+				scheduler.WithGroupedRoundRobinLimits(scheduler.GroupedRoundRobinLimits{
+					MaxSandboxCount: cfg.Scheduler.GroupedRoundRobin.MaxSandboxCount,
+					MaxCPUCount:     cfg.Scheduler.GroupedRoundRobin.MaxCPUCount,
+					MaxMemoryMB:     cfg.Scheduler.GroupedRoundRobin.MaxMemoryMB,
+				}),
+			),
 			store,
 			scheduler.WithArtifactStore(scheduler.NewInMemoryArtifactStore(
 				cfg.Scheduler.ArtifactStoreCapacity,