Verify scoped collection behavior and complete roadmap

2026-10-17 06:39:24 +00:00
parent dfd8da843e
commit 0221cd4610
3 changed files with 138 additions and 36 deletions
--- a/smoke_integration_contract_test.go
+++ b/smoke_integration_contract_test.go
@@ -59,6 +59,89 @@ func TestSmokeTestSharesBackendCapacityWithRunAndReleasesAfterCancellation(t *te
 	}
 }

+func TestScopedCollectionRunsShareBackendCapacity(t *testing.T) {
+	release := make(chan struct{})
+	client := newCapacityGateClient(release, 2)
+	var inputs []configkit.Input
+	for _, namespace := range []string{"north", "south"} {
+		input, err := promptkit.CollectionInput(namespace, configkit.Mapping(namespace+" resources", map[string]any{
+			"profiles": map[string]any{
+				"shared": map[string]any{"backend": "capacity/backend", "model": namespace + "-model"},
+			},
+			"prompts": map[string]any{
+				"shared": map[string]any{
+					"default_profile": namespace + "/shared",
+					"messages":        []any{map[string]any{"role": "user", "content": "Hello"}},
+					"output":          map[string]any{"format": "text", "validation_mode": "none"},
+				},
+			},
+		}))
+		if err != nil {
+			t.Fatal(err)
+		}
+		input.Imports = false
+		inputs = append(inputs, input)
+	}
+	capacityInput, err := promptkit.CollectionInput("capacity", configkit.Mapping("shared capacity", map[string]any{
+		"backends": map[string]any{
+			"backend": map[string]any{
+				"endpoint":          "https://example.invalid/v1",
+				"concurrency_limit": 1,
+				"queue_capacity":    0,
+			},
+		},
+	}))
+	if err != nil {
+		t.Fatal(err)
+	}
+	capacityInput.Imports = false
+	inputs = append(inputs, capacityInput)
+
+	engine, err := promptkit.NewEngine(promptkit.Config{Sources: promptkit.ConfigurationSources{
+		Underlays: []configkit.Underlay{{Inputs: inputs}},
+	}}, promptkit.WithLLMClient(client))
+	if err != nil {
+		t.Fatalf("construct collection engine: %v", err)
+	}
+	south, err := engine.WithNamespaces(promptkit.Namespaces{Prompt: "south", Profile: "south"})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	results := make(chan capacityRunResult, 1)
+	go runCapacityRequest(engine, ctx, promptkit.RunRequest{
+		PromptID: "north/shared",
+	}, results)
+	select {
+	case <-client.started:
+	case <-time.After(5 * time.Second):
+		cancel()
+		t.Fatal("ordinary Run did not reach the shared backend")
+	}
+
+	scopedContext, cancelScoped := context.WithCancel(context.Background())
+	defer cancelScoped()
+	if _, err := south.Run(scopedContext, promptkit.RunRequest{
+		PromptID: "shared", ProfileID: "shared",
+	}); !errors.Is(err, promptkit.ErrCapacityExceeded) {
+		cancel()
+		cancelScoped()
+		t.Fatalf("concurrent scoped Run error = %v, want ErrCapacityExceeded", err)
+	}
+
+	cancel()
+	outcome := awaitCapacityRun(t, results)
+	if outcome.result != nil || !errors.Is(outcome.err, context.Canceled) {
+		t.Fatalf("ordinary Run outcome=(%+v, %v), want context cancellation", outcome.result, outcome.err)
+	}
+	_, peak, calls := client.snapshot()
+	if calls != 1 || peak != 1 {
+		t.Fatalf("shared backend calls=%d peak=%d, want one admitted call", calls, peak)
+	}
+}
+
 type failOnceSmokeClient struct {
 	mu    sync.Mutex
 	calls int