From f310e4ab4ae6e78924c6d37d453f5828bb29a602 Mon Sep 17 00:00:00 2001 From: xu16601526267 <264125260+xu16601526267@users.noreply.github.com> Date: Thu, 6 Aug 2026 13:39:11 +0800 Subject: [PATCH 1/3] fix(runtime): preserve integral config values --- internal/knowledge/configflags.go | 23 ++++++++++++++++++++++- internal/knowledge/configflags_test.go | 18 ++++++++++++++++++ 2 files changed, 40 insertions(+), 1 deletion(-) diff --git a/internal/knowledge/configflags.go b/internal/knowledge/configflags.go index 7ef9ba9f..a2419476 100644 --- a/internal/knowledge/configflags.go +++ b/internal/knowledge/configflags.go @@ -3,8 +3,10 @@ package knowledge import ( "encoding/json" "fmt" + "math" "os" "path/filepath" + "strconv" "strings" ) @@ -15,7 +17,8 @@ import ( // - true bool → "--flag" // - false bool → "--no-flag" // - map / slice → "--flag", -// - other (numbers, strings, etc.) → "--flag", fmt.Sprintf("%v", value) +// - integral floating-point values → "--flag", fixed-point decimal value +// - other (numbers, strings, etc.) → "--flag", fmt.Sprintf("%v", value) // // String template expansion (e.g. {{.ModelPath}}) is the caller's responsibility. func FormatConfigFlag(key string, value any) []string { @@ -31,11 +34,29 @@ func FormatConfigFlag(key string, value any) []string { // YAML-parsed map/slice values are always JSON-marshalable; error is impossible here. b, _ := json.Marshal(value) return []string{flagName, string(b)} + case float64: + return []string{flagName, formatFloatConfigValue(v, 64)} + case float32: + return []string{flagName, formatFloatConfigValue(float64(v), 32)} + case json.Number: + return []string{flagName, v.String()} default: return []string{flagName, fmt.Sprintf("%v", v)} } } +func formatFloatConfigValue(value float64, bitSize int) string { + // JSON/YAML numbers commonly arrive as float64. Go's default %v formatting + // switches sufficiently large values to scientific notation, turning a + // valid CLI value such as 1048576 into 1.048576e+06. Many inference-engine + // integer flags reject that spelling, so preserve integral values as plain + // decimals while retaining compact formatting for real fractions. + if !math.IsNaN(value) && !math.IsInf(value, 0) && math.Trunc(value) == value { + return strconv.FormatFloat(value, 'f', -1, bitSize) + } + return strconv.FormatFloat(value, 'g', -1, bitSize) +} + // ConfigFlagContext describes the runtime command surface used to decide // whether a resolved config key is a real CLI argument or only a resolver hint. type ConfigFlagContext struct { diff --git a/internal/knowledge/configflags_test.go b/internal/knowledge/configflags_test.go index 7c0c3ec6..c459d807 100644 --- a/internal/knowledge/configflags_test.go +++ b/internal/knowledge/configflags_test.go @@ -78,6 +78,24 @@ func TestFormatConfigFlag(t *testing.T) { } }) + t.Run("integral JSON numbers stay in decimal notation", func(t *testing.T) { + cases := []struct { + value any + want string + }{ + {float64(1048576), "1048576"}, + {float32(1048576), "1048576"}, + {json.Number("1048576"), "1048576"}, + {float64(0.835), "0.835"}, + } + for _, tc := range cases { + got := FormatConfigFlag("max_model_len", tc.value) + if len(got) != 2 || got[1] != tc.want { + t.Fatalf("value=%v: got %v, want decimal value %q", tc.value, got, tc.want) + } + } + }) + t.Run("underscore keys become hyphenated flags", func(t *testing.T) { got := FormatConfigFlag("mem_fraction_static", 0.7) if got[0] != "--mem-fraction-static" { From b7d803fde2b9e8907e7749560899cb87d7a28e4d Mon Sep 17 00:00:00 2001 From: xu16601526267 <264125260+xu16601526267@users.noreply.github.com> Date: Thu, 6 Aug 2026 13:39:17 +0800 Subject: [PATCH 2/3] feat(scenario): orchestrate deployments across Fleet devices --- cmd/aima/main_test.go | 85 +++++++++- cmd/aima/scenario.go | 251 ++++++++++++++++++++++++++++-- cmd/aima/tooldeps_deploy.go | 68 ++++++++ cmd/aima/tooldeps_deploy_test.go | 20 +++ cmd/aima/tooldeps_integration.go | 5 +- internal/cli/scenario.go | 12 +- internal/knowledge/loader.go | 72 +++++---- internal/knowledge/podgen.go | 12 +- internal/knowledge/podgen_test.go | 16 ++ internal/knowledge/resolver.go | 64 +++++++- internal/mcp/tools_deps.go | 2 +- internal/mcp/tools_scenario.go | 12 +- internal/runtime/docker.go | 11 ++ internal/runtime/docker_test.go | 2 + 14 files changed, 574 insertions(+), 58 deletions(-) diff --git a/cmd/aima/main_test.go b/cmd/aima/main_test.go index 89c84a4d..650e81c4 100644 --- a/cmd/aima/main_test.go +++ b/cmd/aima/main_test.go @@ -1190,7 +1190,7 @@ func TestApplyScenarioSkipsRemainingDeploymentsAndPostDeployAfterWaitFailure(t * }, } - data, err := applyScenario(context.Background(), cat, "docker", deps, "demo", false) + data, err := applyScenario(context.Background(), cat, "docker", deps, "demo", false, nil) if err != nil { t.Fatalf("applyScenario: %v", err) } @@ -1251,7 +1251,7 @@ func TestApplyScenarioWaitsOnLastStepBeforePostDeploy(t *testing.T) { }, } - data, err := applyScenario(context.Background(), cat, "docker", deps, "demo-last", false) + data, err := applyScenario(context.Background(), cat, "docker", deps, "demo-last", false, nil) if err != nil { t.Fatalf("applyScenario: %v", err) } @@ -1276,6 +1276,87 @@ func TestApplyScenarioWaitsOnLastStepBeforePostDeploy(t *testing.T) { } } +func TestApplyScenarioDryRunRoutesRemoteDeploymentAndExpandsBindings(t *testing.T) { + cat := &knowledge.Catalog{DeploymentScenarios: []knowledge.DeploymentScenario{{ + Metadata: knowledge.ScenarioMetadata{Name: "two-node"}, + Inputs: []knowledge.ScenarioInput{ + {Name: "worker_device", Required: true}, + {Name: "master_addr", Required: true}, + }, + Deployments: []knowledge.ScenarioDeployment{ + {ID: "worker", Device: "{{.worker_device}}", Model: "same-model", Engine: "engine", Config: map[string]any{"node_rank": 1}, Env: map[string]string{"MASTER_ADDR": "{{.master_addr}}"}}, + {ID: "head", Device: "local", Model: "same-model", Engine: "engine", Config: map[string]any{"node_rank": 0}}, + }, + StartupOrder: []knowledge.ScenarioStartupStep{ + {Step: 1, Deployment: "worker"}, + {Step: 2, Deployment: "head"}, + }, + }}} + + var remoteParams map[string]any + localCalls := 0 + deps := &mcp.ToolDeps{ + FleetExecTool: func(ctx context.Context, deviceID, toolName string, params json.RawMessage) (json.RawMessage, error) { + if deviceID != "spark-worker" || toolName != "deploy.dry_run" { + t.Fatalf("unexpected remote call device=%q tool=%q", deviceID, toolName) + } + if err := json.Unmarshal(params, &remoteParams); err != nil { + t.Fatal(err) + } + return json.Marshal(mcp.TextResult(`{"name":"remote-plan"}`)) + }, + DeployDryRun: func(ctx context.Context, engine, model, slot string, config map[string]any) (json.RawMessage, error) { + localCalls++ + if rank := config["node_rank"]; rank != 0 { + t.Fatalf("local node_rank = %#v, want 0", rank) + } + return json.RawMessage(`{"name":"local-plan"}`), nil + }, + } + + data, err := applyScenario(context.Background(), cat, "docker", deps, "two-node", true, map[string]string{ + "worker_device": "spark-worker", + "master_addr": "10.0.0.1", + }) + if err != nil { + t.Fatalf("applyScenario: %v", err) + } + if localCalls != 1 { + t.Fatalf("local dry-run calls = %d, want 1", localCalls) + } + config, ok := remoteParams["config"].(map[string]any) + if !ok { + t.Fatalf("remote params = %#v, config missing", remoteParams) + } + env, ok := config["_env"].(map[string]any) + if !ok { + t.Fatalf("remote config = %#v, _env missing", config) + } + if env["MASTER_ADDR"] != "10.0.0.1" { + t.Fatalf("remote env = %#v", env) + } + var response struct { + Deployments []scenarioDeployResult `json:"deployments"` + } + if err := json.Unmarshal(data, &response); err != nil { + t.Fatal(err) + } + if len(response.Deployments) != 2 || response.Deployments[0].Device != "spark-worker" || response.Deployments[1].Device != "local" { + t.Fatalf("deployments = %#v", response.Deployments) + } +} + +func TestApplyScenarioRejectsMissingRequiredBinding(t *testing.T) { + cat := &knowledge.Catalog{DeploymentScenarios: []knowledge.DeploymentScenario{{ + Metadata: knowledge.ScenarioMetadata{Name: "needs-input"}, + Inputs: []knowledge.ScenarioInput{{Name: "worker_device", Required: true}}, + }}} + _, err := applyScenario(context.Background(), cat, "docker", &mcp.ToolDeps{}, "needs-input", true, nil) + if err == nil || !strings.Contains(err.Error(), "worker_device") { + t.Fatalf("error = %v, want missing worker_device", err) + } +} + func TestVariantQuantizationHint(t *testing.T) { if got := variantQuantizationHint(&knowledge.ModelVariant{ DefaultConfig: map[string]any{"quantization": "gptq"}, diff --git a/cmd/aima/scenario.go b/cmd/aima/scenario.go index ea495a13..99704676 100644 --- a/cmd/aima/scenario.go +++ b/cmd/aima/scenario.go @@ -6,6 +6,7 @@ import ( "fmt" "log/slog" "net" + "regexp" "sort" "strings" "time" @@ -17,6 +18,7 @@ import ( type scenarioDeployResult struct { Model string `json:"model"` Engine string `json:"engine"` + Device string `json:"device,omitempty"` Status string `json:"status"` Error string `json:"error,omitempty"` Data json.RawMessage `json:"data,omitempty"` @@ -28,7 +30,7 @@ type orderedScenarioDeploy struct { timeoutS int } -func applyScenario(ctx context.Context, cat *knowledge.Catalog, rtName string, deps *mcp.ToolDeps, name string, dryRun bool) (json.RawMessage, error) { +func applyScenario(ctx context.Context, cat *knowledge.Catalog, rtName string, deps *mcp.ToolDeps, name string, dryRun bool, provided map[string]string) (json.RawMessage, error) { var scenario *knowledge.DeploymentScenario for i := range cat.DeploymentScenarios { if strings.EqualFold(cat.DeploymentScenarios[i].Metadata.Name, name) { @@ -43,6 +45,10 @@ func applyScenario(ctx context.Context, cat *knowledge.Catalog, rtName string, d } return nil, fmt.Errorf("scenario %q not found (available: %v)", name, names) } + bindings, err := resolveScenarioBindings(scenario.Inputs, provided) + if err != nil { + return nil, err + } var results []scenarioDeployResult @@ -58,32 +64,41 @@ func applyScenario(ctx context.Context, cat *knowledge.Catalog, rtName string, d var ordered []orderedScenarioDeploy if len(scenario.StartupOrder) > 0 { - byModel := make(map[string]knowledge.ScenarioDeployment, len(scenario.Deployments)) - for _, d := range scenario.Deployments { - byModel[strings.ToLower(d.Model)] = d + byKey := make(map[string]int, len(scenario.Deployments)) + for i, d := range scenario.Deployments { + key := scenarioDeploymentKey(d) + if _, exists := byKey[key]; exists { + return nil, fmt.Errorf("scenario %q has duplicate deployment key %q; set unique deployment ids", name, key) + } + byKey[key] = i } + used := make(map[int]bool, len(scenario.Deployments)) steps := make([]knowledge.ScenarioStartupStep, len(scenario.StartupOrder)) copy(steps, scenario.StartupOrder) sort.Slice(steps, func(i, j int) bool { return steps[i].Step < steps[j].Step }) for _, step := range steps { - d, ok := byModel[strings.ToLower(step.Model)] + ref := step.Deployment + if ref == "" { + ref = step.Model + } + idx, ok := byKey[strings.ToLower(strings.TrimSpace(ref))] if !ok { results = append(results, scenarioDeployResult{ - Model: step.Model, + Model: ref, Status: "error", - Error: fmt.Sprintf("startup_order references unknown model %q", step.Model), + Error: fmt.Sprintf("startup_order references unknown deployment %q", ref), }) continue } ordered = append(ordered, orderedScenarioDeploy{ - deployment: d, + deployment: scenario.Deployments[idx], waitFor: step.WaitFor, timeoutS: step.TimeoutS, }) - delete(byModel, strings.ToLower(step.Model)) + used[idx] = true } - for _, d := range scenario.Deployments { - if _, remaining := byModel[strings.ToLower(d.Model)]; remaining { + for i, d := range scenario.Deployments { + if !used[i] { ordered = append(ordered, orderedScenarioDeploy{deployment: d}) } } @@ -96,18 +111,32 @@ func applyScenario(ctx context.Context, cat *knowledge.Catalog, rtName string, d blockFurther := false blockReason := "" for i, od := range ordered { - d := od.deployment + d, err := resolveScenarioDeployment(od.deployment, bindings) + if err != nil { + return nil, fmt.Errorf("resolve deployment %q: %w", scenarioDeploymentKey(od.deployment), err) + } + device := strings.TrimSpace(d.Device) + remote := device != "" && !strings.EqualFold(device, "local") + config := cloneScenarioConfig(d.Config) + if len(d.Env) > 0 { + env := make(map[string]any, len(d.Env)) + for key, value := range d.Env { + env[key] = value + } + config["_env"] = env + } if blockFurther && !dryRun { results = append(results, scenarioDeployResult{ Model: d.Model, Engine: d.Engine, + Device: device, Status: "skipped", Error: fmt.Sprintf("skipped after earlier deployment failure: %s", blockReason), }) continue } if dryRun { - if deps.DeployDryRun == nil { + if (!remote && deps.DeployDryRun == nil) || (remote && deps.FleetExecTool == nil) { results = append(results, scenarioDeployResult{ Model: d.Model, Engine: d.Engine, @@ -116,11 +145,17 @@ func applyScenario(ctx context.Context, cat *knowledge.Catalog, rtName string, d }) continue } - data, err := deps.DeployDryRun(ctx, d.Engine, d.Model, d.Slot, d.Config) + var data json.RawMessage + if remote { + data, err = scenarioFleetDeploy(ctx, deps, device, "deploy.dry_run", d, config) + } else { + data, err = deps.DeployDryRun(ctx, d.Engine, d.Model, d.Slot, config) + } if err != nil { results = append(results, scenarioDeployResult{ Model: d.Model, Engine: d.Engine, + Device: device, Status: "error", Error: err.Error(), }) @@ -128,6 +163,7 @@ func applyScenario(ctx context.Context, cat *knowledge.Catalog, rtName string, d results = append(results, scenarioDeployResult{ Model: d.Model, Engine: d.Engine, + Device: device, Status: "dry_run", Data: data, }) @@ -135,7 +171,7 @@ func applyScenario(ctx context.Context, cat *knowledge.Catalog, rtName string, d continue } - if deps.DeployApply == nil { + if (!remote && deps.DeployApply == nil) || (remote && deps.FleetExecTool == nil) { blockFurther = true blockReason = "deploy.apply not available" results = append(results, scenarioDeployResult{ @@ -146,7 +182,12 @@ func applyScenario(ctx context.Context, cat *knowledge.Catalog, rtName string, d }) continue } - data, err := deps.DeployApply(ctx, d.Engine, d.Model, d.Slot, d.Config, false) + var data json.RawMessage + if remote { + data, err = scenarioFleetDeploy(ctx, deps, device, "deploy.apply", d, config) + } else { + data, err = deps.DeployApply(ctx, d.Engine, d.Model, d.Slot, config, d.NoPull) + } if err != nil { blockFurther = true blockReason = err.Error() @@ -162,6 +203,7 @@ func applyScenario(ctx context.Context, cat *knowledge.Catalog, rtName string, d results = append(results, scenarioDeployResult{ Model: d.Model, Engine: d.Engine, + Device: device, Status: "ok", Data: data, }) @@ -176,7 +218,18 @@ func applyScenario(ctx context.Context, cat *knowledge.Catalog, rtName string, d shouldWait := i < len(ordered)-1 || od.waitFor != "" || od.timeoutS > 0 if shouldWait { - if err := scenarioWaitForReady(ctx, deploymentQuery, od.waitFor, od.timeoutS, deps.DeployStatus); err != nil { + deployStatus := deps.DeployStatus + if remote { + deployStatus = func(waitCtx context.Context, query string) (json.RawMessage, error) { + params, _ := json.Marshal(map[string]any{"name": query}) + raw, callErr := deps.FleetExecTool(waitCtx, device, "deploy.status", params) + if callErr != nil { + return nil, callErr + } + return unwrapFleetToolResult(raw) + } + } + if err := scenarioWaitForReady(ctx, deploymentQuery, od.waitFor, od.timeoutS, deployStatus); err != nil { slog.Warn("startup wait did not complete", "model", d.Model, "wait_for", od.waitFor, "err", err) blockFurther = true blockReason = err.Error() @@ -246,6 +299,170 @@ func applyScenario(ctx context.Context, cat *knowledge.Catalog, rtName string, d return json.Marshal(resp) } +var scenarioBindingPattern = regexp.MustCompile(`\{\{\.([A-Za-z0-9_-]+)\}\}`) + +func scenarioDeploymentKey(d knowledge.ScenarioDeployment) string { + if strings.TrimSpace(d.ID) != "" { + return strings.ToLower(strings.TrimSpace(d.ID)) + } + return strings.ToLower(strings.TrimSpace(d.Model)) +} + +func resolveScenarioBindings(inputs []knowledge.ScenarioInput, provided map[string]string) (map[string]string, error) { + bindings := make(map[string]string, len(inputs)+len(provided)) + for _, input := range inputs { + if input.Default != "" { + bindings[input.Name] = input.Default + } + } + for key, value := range provided { + bindings[key] = value + } + var missing []string + for _, input := range inputs { + if input.Required && strings.TrimSpace(bindings[input.Name]) == "" { + missing = append(missing, input.Name) + } + } + if len(missing) > 0 { + sort.Strings(missing) + return nil, fmt.Errorf("scenario requires bindings: %s", strings.Join(missing, ", ")) + } + return bindings, nil +} + +func resolveScenarioDeployment(d knowledge.ScenarioDeployment, bindings map[string]string) (knowledge.ScenarioDeployment, error) { + var err error + expand := func(value string) string { + if err != nil { + return value + } + value, err = expandScenarioString(value, bindings) + return value + } + d.Device = expand(d.Device) + d.Model = expand(d.Model) + d.Engine = expand(d.Engine) + d.Slot = expand(d.Slot) + d.Notes = expand(d.Notes) + if err != nil { + return d, err + } + resolved, err := expandScenarioValue(d.Config, bindings) + if err != nil { + return d, err + } + if resolved != nil { + d.Config = resolved.(map[string]any) + } + originalEnv := d.Env + d.Env = make(map[string]string, len(originalEnv)) + for key, value := range originalEnv { + d.Env[key], err = expandScenarioString(value, bindings) + if err != nil { + return d, err + } + } + return d, nil +} + +func expandScenarioString(value string, bindings map[string]string) (string, error) { + value = scenarioBindingPattern.ReplaceAllStringFunc(value, func(token string) string { + match := scenarioBindingPattern.FindStringSubmatch(token) + if len(match) != 2 { + return token + } + if replacement, ok := bindings[match[1]]; ok { + return replacement + } + return token + }) + if match := scenarioBindingPattern.FindStringSubmatch(value); len(match) == 2 { + return "", fmt.Errorf("missing binding %q", match[1]) + } + return value, nil +} + +func expandScenarioValue(value any, bindings map[string]string) (any, error) { + switch typed := value.(type) { + case string: + return expandScenarioString(typed, bindings) + case map[string]any: + out := make(map[string]any, len(typed)) + for key, child := range typed { + expanded, err := expandScenarioValue(child, bindings) + if err != nil { + return nil, err + } + out[key] = expanded + } + return out, nil + case []any: + out := make([]any, len(typed)) + for i, child := range typed { + expanded, err := expandScenarioValue(child, bindings) + if err != nil { + return nil, err + } + out[i] = expanded + } + return out, nil + default: + return value, nil + } +} + +func cloneScenarioConfig(config map[string]any) map[string]any { + out := make(map[string]any, len(config)+1) + for key, value := range config { + out[key] = value + } + return out +} + +func scenarioFleetDeploy(ctx context.Context, deps *mcp.ToolDeps, device, toolName string, d knowledge.ScenarioDeployment, config map[string]any) (json.RawMessage, error) { + if deps.FleetExecTool == nil { + return nil, fmt.Errorf("fleet.exec not available for remote device %q", device) + } + params, err := json.Marshal(map[string]any{ + "model": d.Model, + "engine": d.Engine, + "slot": d.Slot, + "config": config, + "no_pull": d.NoPull, + }) + if err != nil { + return nil, err + } + raw, err := deps.FleetExecTool(ctx, device, toolName, params) + if err != nil { + return nil, err + } + return unwrapFleetToolResult(raw) +} + +func unwrapFleetToolResult(raw json.RawMessage) (json.RawMessage, error) { + var result mcp.ToolResult + if err := json.Unmarshal(raw, &result); err != nil || len(result.Content) == 0 { + return raw, nil + } + var textParts []string + for _, content := range result.Content { + if content.Type == "text" { + textParts = append(textParts, content.Text) + } + } + text := strings.Join(textParts, "\n") + if result.IsError { + return nil, fmt.Errorf("remote tool failed: %s", text) + } + if json.Valid([]byte(text)) { + return json.RawMessage(text), nil + } + encoded, err := json.Marshal(text) + return encoded, err +} + // scenarioWaitForReady waits for a deployed model to become ready before proceeding. // waitFor: "health_check" polls deploy.status, "port_open" probes the returned address, "" defaults to 2s sleep. // On timeout, returns an error (caller treats as warning, continues deployment). diff --git a/cmd/aima/tooldeps_deploy.go b/cmd/aima/tooldeps_deploy.go index 3319fb67..953342bb 100644 --- a/cmd/aima/tooldeps_deploy.go +++ b/cmd/aima/tooldeps_deploy.go @@ -43,6 +43,10 @@ func buildDeployDeps(ac *appContext, deps *mcp.ToolDeps, dataDir := ac.dataDir deps.DeployApply = func(ctx context.Context, engineType, modelName, slot string, configOverrides map[string]any, noPull bool) (json.RawMessage, error) { + configOverrides, envOverrides, err := splitDeploymentEnvOverrides(configOverrides) + if err != nil { + return nil, err + } if noPull { ctx = withDeployAutoPull(ctx, false) } @@ -67,6 +71,16 @@ func buildDeployDeps(ac *appContext, deps *mcp.ToolDeps, } modelName = rd.ModelName resolved := rd.Resolved + if len(envOverrides) > 0 { + mergedEnv := make(map[string]string, len(resolved.Env)+len(envOverrides)) + for key, value := range resolved.Env { + mergedEnv[key] = value + } + for key, value := range envOverrides { + mergedEnv[key] = value + } + resolved.Env = mergedEnv + } upstreamModel := resolvedServedModelName(modelName, resolved.Config) modelPath, modelPathErr := resolveLocalModelPathNoPull(modelName, resolved, dataDir) @@ -342,6 +356,12 @@ func buildDeployDeps(ac *appContext, deps *mcp.ToolDeps, } deps.DeployDryRun = func(ctx context.Context, engineType, modelName, slot string, overrides map[string]any) (json.RawMessage, error) { + var envOverrides map[string]string + var err error + overrides, envOverrides, err = splitDeploymentEnvOverrides(overrides) + if err != nil { + return nil, err + } hwInfo := buildHardwareInfo(ctx, cat, rt.Name()) rd, err := resolveDeployment(ctx, cat, db, kStore, hwInfo, modelName, engineType, slot, overrides, dataDir) if err != nil { @@ -350,6 +370,9 @@ func buildDeployDeps(ac *appContext, deps *mcp.ToolDeps, // Select runtime for display resolved := rd.Resolved + if len(envOverrides) > 0 { + resolved.Env = mergeDeploymentEnv(resolved.Env, envOverrides) + } hasPartition := resolved.Partition != nil && (resolved.Partition.GPUMemoryMiB > 0 || resolved.Partition.GPUCoresPercent > 0) selectedRt, rtErr := pickRuntimeForDeployment(resolved.RuntimeRecommendation, k3sRt, dockerRt, nativeRt, rt, hasPartition) if rtErr != nil { @@ -380,6 +403,7 @@ func buildDeployDeps(ac *appContext, deps *mcp.ToolDeps, "config": resolved.Config, "resolved_config": rd.ResolvedConfig, "effective_config": resolved.Config, + "env": resolved.Env, "fit_adjustments": rd.Fit.Adjustments, "ports": knowledge.ResolvePortBindingsFromSpecs(resolved.PortSpecs, resolved.Config), "provenance": resolved.Provenance, @@ -634,6 +658,50 @@ func buildDeployDeps(ac *appContext, deps *mcp.ToolDeps, } } +func splitDeploymentEnvOverrides(config map[string]any) (map[string]any, map[string]string, error) { + if len(config) == 0 { + return config, nil, nil + } + clean := make(map[string]any, len(config)) + for key, value := range config { + if key != "_env" { + clean[key] = value + } + } + raw, ok := config["_env"] + if !ok || raw == nil { + return clean, nil, nil + } + env := map[string]string{} + switch values := raw.(type) { + case map[string]string: + for key, value := range values { + env[key] = value + } + case map[string]any: + for key, value := range values { + if strings.TrimSpace(key) == "" { + return nil, nil, fmt.Errorf("_env contains an empty variable name") + } + env[key] = fmt.Sprint(value) + } + default: + return nil, nil, fmt.Errorf("_env must be an object of environment variables") + } + return clean, env, nil +} + +func mergeDeploymentEnv(base, overrides map[string]string) map[string]string { + merged := make(map[string]string, len(base)+len(overrides)) + for key, value := range base { + merged[key] = value + } + for key, value := range overrides { + merged[key] = value + } + return merged +} + func setActiveLLMModelConfigForType(ctx context.Context, db *state.DB, modelName, modelType string) error { modelName = strings.TrimSpace(modelName) if db == nil || modelName == "" { diff --git a/cmd/aima/tooldeps_deploy_test.go b/cmd/aima/tooldeps_deploy_test.go index ef14802c..8da4742d 100644 --- a/cmd/aima/tooldeps_deploy_test.go +++ b/cmd/aima/tooldeps_deploy_test.go @@ -1,6 +1,7 @@ package main import ( + "reflect" "testing" "github.com/jguan/aima/internal/knowledge" @@ -8,6 +9,25 @@ import ( "github.com/jguan/aima/internal/runtime" ) +func TestSplitDeploymentEnvOverrides(t *testing.T) { + config, env, err := splitDeploymentEnvOverrides(map[string]any{ + "max_model_len": 1048576, + "_env": map[string]any{ + "NODE_RANK": 1, + "HEADLESS": true, + }, + }) + if err != nil { + t.Fatalf("splitDeploymentEnvOverrides: %v", err) + } + if !reflect.DeepEqual(config, map[string]any{"max_model_len": 1048576}) { + t.Fatalf("config = %#v", config) + } + if !reflect.DeepEqual(env, map[string]string{"NODE_RANK": "1", "HEADLESS": "true"}) { + t.Fatalf("env = %#v", env) + } +} + func TestResolvedServedModelNameExpandsModelTemplate(t *testing.T) { got := resolvedServedModelName("GLM-4.1V-9B-Thinking-FP4", map[string]any{ "served_model_name": "{{.ModelName}}", diff --git a/cmd/aima/tooldeps_integration.go b/cmd/aima/tooldeps_integration.go index 12835abb..87359c4b 100644 --- a/cmd/aima/tooldeps_integration.go +++ b/cmd/aima/tooldeps_integration.go @@ -85,6 +85,7 @@ func buildIntegrationDeps(ac *appContext, deps *mcp.ToolDeps) { "memory_budget": ds.MemoryBudget, "startup_order": ds.StartupOrder, "alternative_configs": ds.AlternativeConfigs, + "inputs": ds.Inputs, }) } } @@ -95,8 +96,8 @@ func buildIntegrationDeps(ac *appContext, deps *mcp.ToolDeps) { return nil, fmt.Errorf("scenario %q not found (available: %v)", name, names) } - deps.ScenarioApply = func(ctx context.Context, name string, dryRun bool) (json.RawMessage, error) { - return applyScenario(ctx, cat, ac.rt.Name(), deps, name, dryRun) + deps.ScenarioApply = func(ctx context.Context, name string, dryRun bool, bindings map[string]string) (json.RawMessage, error) { + return applyScenario(ctx, cat, ac.rt.Name(), deps, name, dryRun, bindings) } // Knowledge sync (K6) diff --git a/internal/cli/scenario.go b/internal/cli/scenario.go index 6077142c..fe3e891e 100644 --- a/internal/cli/scenario.go +++ b/internal/cli/scenario.go @@ -64,6 +64,7 @@ func newScenarioShowCmd(app *App) *cobra.Command { func newScenarioApplyCmd(app *App) *cobra.Command { var dryRun bool + var bindingValues []string cmd := &cobra.Command{ Use: "apply ", Short: "Deploy all models defined in a scenario", @@ -72,7 +73,15 @@ func newScenarioApplyCmd(app *App) *cobra.Command { if app.ToolDeps == nil || app.ToolDeps.ScenarioApply == nil { return fmt.Errorf("scenario.apply not available") } - data, err := app.ToolDeps.ScenarioApply(cmd.Context(), args[0], dryRun) + bindings := make(map[string]string, len(bindingValues)) + for _, item := range bindingValues { + key, value, ok := strings.Cut(item, "=") + if !ok || strings.TrimSpace(key) == "" { + return fmt.Errorf("invalid --set %q: expected key=value", item) + } + bindings[strings.TrimSpace(key)] = value + } + data, err := app.ToolDeps.ScenarioApply(cmd.Context(), args[0], dryRun, bindings) if err != nil { return err } @@ -81,6 +90,7 @@ func newScenarioApplyCmd(app *App) *cobra.Command { }, } cmd.Flags().BoolVar(&dryRun, "dry-run", false, "Preview deployments without executing") + cmd.Flags().StringArrayVar(&bindingValues, "set", nil, "Set a scenario input (repeatable key=value)") return cmd } diff --git a/internal/knowledge/loader.go b/internal/knowledge/loader.go index 45256e27..d1714bd3 100644 --- a/internal/knowledge/loader.go +++ b/internal/knowledge/loader.go @@ -103,18 +103,19 @@ type HardwarePartition struct { } // ContainerAccess describes vendor-specific container access requirements -// (devices, env vars, volumes, security) for GPU containers. Lives in -// hardware profile YAML so adding a new GPU vendor = YAML only, no Go code. +// (devices, env vars, volumes, security) for GPU containers. Hardware profiles +// provide vendor defaults; engine assets may add engine-specific requirements. type ContainerAccess struct { - Devices []string `yaml:"devices,omitempty"` - Env map[string]string `yaml:"env,omitempty"` - PartitionRemoveEnv []string `yaml:"partition_remove_env,omitempty"` - Volumes []ContainerVolume `yaml:"volumes,omitempty"` - Security *ContainerSecurity `yaml:"security,omitempty"` - DockerRuntime string `yaml:"docker_runtime,omitempty"` // --runtime flag (e.g. "ascend") - NetworkMode string `yaml:"network_mode,omitempty"` // "host" for --network host - ShmSize string `yaml:"shm_size,omitempty"` // --shm-size (e.g. "500g") - Init bool `yaml:"init,omitempty"` // --init flag + Devices []string `yaml:"devices,omitempty" json:"devices,omitempty"` + Env map[string]string `yaml:"env,omitempty" json:"env,omitempty"` + Ulimits map[string]string `yaml:"ulimits,omitempty" json:"ulimits,omitempty"` + PartitionRemoveEnv []string `yaml:"partition_remove_env,omitempty" json:"partition_remove_env,omitempty"` + Volumes []ContainerVolume `yaml:"volumes,omitempty" json:"volumes,omitempty"` + Security *ContainerSecurity `yaml:"security,omitempty" json:"security,omitempty"` + DockerRuntime string `yaml:"docker_runtime,omitempty" json:"docker_runtime,omitempty"` // --runtime flag (e.g. "ascend") + NetworkMode string `yaml:"network_mode,omitempty" json:"network_mode,omitempty"` // "host" for --network host + ShmSize string `yaml:"shm_size,omitempty" json:"shm_size,omitempty"` // --shm-size (e.g. "500g") + Init bool `yaml:"init,omitempty" json:"init,omitempty"` // --init flag } type ContainerVolume struct { @@ -125,9 +126,9 @@ type ContainerVolume struct { } type ContainerSecurity struct { - Privileged bool `yaml:"privileged,omitempty"` - RunAsUser *int `yaml:"run_as_user,omitempty"` - SupplementalGroups []int `yaml:"supplemental_groups,omitempty"` + Privileged bool `yaml:"privileged,omitempty" json:"privileged,omitempty"` + RunAsUser *int `yaml:"run_as_user,omitempty" json:"run_as_user,omitempty"` + SupplementalGroups []int `yaml:"supplemental_groups,omitempty" json:"supplemental_groups,omitempty"` } // --- Engine Asset --- @@ -191,6 +192,7 @@ type EngineAsset struct { Runtime EngineRuntime `yaml:"runtime,omitempty" json:"runtime,omitempty"` Patterns []string `yaml:"patterns,omitempty" json:"patterns,omitempty"` Source *EngineSource `yaml:"source,omitempty" json:"source,omitempty"` + Container *ContainerAccess `yaml:"container,omitempty" json:"container,omitempty"` OpenQuestions []StackQuestion `yaml:"open_questions,omitempty" json:"open_questions,omitempty"` } @@ -352,7 +354,8 @@ type ModelUI struct { } type ModelCapabilities struct { - StandaloneDeploy *bool `yaml:"standalone_deploy,omitempty"` + StandaloneDeploy *bool `yaml:"standalone_deploy,omitempty"` + DeploymentScenario string `yaml:"deployment_scenario,omitempty"` } type OpenClawHints struct { @@ -611,6 +614,16 @@ type DeploymentScenario struct { MemoryBudget map[string]any `yaml:"memory_budget,omitempty"` StartupOrder []ScenarioStartupStep `yaml:"startup_order,omitempty"` AlternativeConfigs []ScenarioAlternative `yaml:"alternative_configs,omitempty"` + Inputs []ScenarioInput `yaml:"inputs,omitempty"` +} + +type ScenarioInput struct { + Name string `yaml:"name" json:"name"` + Label string `yaml:"label,omitempty" json:"label,omitempty"` + Description string `yaml:"description,omitempty" json:"description,omitempty"` + Kind string `yaml:"kind,omitempty" json:"kind,omitempty"` + Default string `yaml:"default,omitempty" json:"default,omitempty"` + Required bool `yaml:"required,omitempty" json:"required,omitempty"` } type ScenarioMetadata struct { @@ -624,13 +637,17 @@ type ScenarioTarget struct { } type ScenarioDeployment struct { - Model string `yaml:"model"` - Engine string `yaml:"engine"` - Slot string `yaml:"slot,omitempty"` - Role string `yaml:"role,omitempty"` - Modalities []string `yaml:"modalities,omitempty"` - Config map[string]any `yaml:"config,omitempty"` - Notes string `yaml:"notes,omitempty"` + ID string `yaml:"id,omitempty"` + Device string `yaml:"device,omitempty"` + Model string `yaml:"model"` + Engine string `yaml:"engine"` + Slot string `yaml:"slot,omitempty"` + Role string `yaml:"role,omitempty"` + Modalities []string `yaml:"modalities,omitempty"` + Config map[string]any `yaml:"config,omitempty"` + Env map[string]string `yaml:"env,omitempty"` + NoPull bool `yaml:"no_pull,omitempty"` + Notes string `yaml:"notes,omitempty"` } type ScenarioAction struct { @@ -646,11 +663,12 @@ type ScenarioVerification struct { } type ScenarioStartupStep struct { - Step int `yaml:"step"` - Model string `yaml:"model"` - WaitFor string `yaml:"wait_for"` - TimeoutS int `yaml:"timeout_s"` - Notes string `yaml:"notes,omitempty"` + Step int `yaml:"step"` + Deployment string `yaml:"deployment,omitempty"` + Model string `yaml:"model"` + WaitFor string `yaml:"wait_for"` + TimeoutS int `yaml:"timeout_s"` + Notes string `yaml:"notes,omitempty"` } type ScenarioAlternative struct { diff --git a/internal/knowledge/podgen.go b/internal/knowledge/podgen.go index b26d7dfd..234550a5 100644 --- a/internal/knowledge/podgen.go +++ b/internal/knowledge/podgen.go @@ -5,6 +5,7 @@ import ( "fmt" "path/filepath" "sort" + "strconv" "strings" "text/template" @@ -14,6 +15,7 @@ import ( var podTemplate = template.Must(template.New("pod").Funcs(template.FuncMap{ "deviceVolName": deviceVolName, "containerPortName": containerPortName, + "yamlQuote": strconv.Quote, }).Parse(`apiVersion: v1 kind: Pod metadata: @@ -35,6 +37,10 @@ metadata: spec: schedulerName: default-scheduler restartPolicy: Always + {{- if .HostNetwork }} + hostNetwork: true + dnsPolicy: ClusterFirstWithHostNet + {{- end }} {{- if .RuntimeClassName }} runtimeClassName: {{ .RuntimeClassName }} {{- end }} @@ -54,7 +60,7 @@ spec: {{- if .Args }} command: {{- range .Args }} - - "{{ . }}" + - {{ yamlQuote . }} {{- end }} {{- end }} {{- if .Ports }} @@ -68,7 +74,7 @@ spec: env: {{- range $k, $v := .ExtraEnv }} - name: {{ $k }} - value: "{{ $v }}" + value: {{ yamlQuote $v }} {{- end }} {{- end }} {{- if .HasContainerSecurity }} @@ -187,6 +193,7 @@ type podData struct { Devices []string // device paths to mount, e.g. ["/dev/kfd", "/dev/dri"] ExtraVolumes []ContainerVolume // additional host mounts Security *ContainerSecurity // pod-level securityContext + HostNetwork bool } func (d podData) HasAnnotations() bool { @@ -364,6 +371,7 @@ func GeneratePod(resolved *ResolvedConfig) ([]byte, error) { data.Devices = resolved.Container.Devices data.ExtraVolumes = resolved.Container.Volumes data.Security = resolved.Container.Security + data.HostNetwork = resolved.Container.NetworkMode == "host" } // Merge engine extra_volumes (e.g. patch scripts) into pod volumes. diff --git a/internal/knowledge/podgen_test.go b/internal/knowledge/podgen_test.go index 340c5802..05283ff9 100644 --- a/internal/knowledge/podgen_test.go +++ b/internal/knowledge/podgen_test.go @@ -225,6 +225,22 @@ func TestGeneratePodNilResolved(t *testing.T) { } } +func TestGeneratePodHostNetwork(t *testing.T) { + resolved := &ResolvedConfig{ + Engine: "distributed-vllm", EngineImage: "example/vllm:latest", ModelName: "model", + ModelPath: "/models/model", Command: []string{"vllm", "serve", "{{.ModelPath}}"}, + Container: &ContainerAccess{NetworkMode: "host"}, + } + podYAML, err := GeneratePod(resolved) + if err != nil { + t.Fatalf("GeneratePod: %v", err) + } + text := string(podYAML) + if !strings.Contains(text, "hostNetwork: true") || !strings.Contains(text, "dnsPolicy: ClusterFirstWithHostNet") { + t.Fatalf("host-network fields missing:\n%s", text) + } +} + func TestGeneratePodMemGuardrail(t *testing.T) { // No partition: the resolver's MemLimitMiB guardrail (unified-memory hosts) // must still produce a container memory ceiling so a runaway pod is diff --git a/internal/knowledge/resolver.go b/internal/knowledge/resolver.go index 906a61c0..bb4417a9 100644 --- a/internal/knowledge/resolver.go +++ b/internal/knowledge/resolver.go @@ -256,7 +256,7 @@ func (c *Catalog) Resolve(hw HardwareInfo, modelName, engineType string, userOve resolved.GPUResourceName = c.findGPUResourceName(hw) resolved.RuntimeClassName = c.findRuntimeClassName(hw) resolved.CPUArch = hw.CPUArch - resolved.Container = c.findContainerAccess(hw) + resolved.Container = mergeContainerAccess(c.findContainerAccess(hw), engine.Container) // Set runtime recommendation from engine's platform_recommendations if rec, ok := engine.Runtime.PlatformRecommendations[hw.Platform]; ok { @@ -774,6 +774,68 @@ func (c *Catalog) findContainerAccess(hw HardwareInfo) *ContainerAccess { return nil } +func mergeContainerAccess(base, override *ContainerAccess) *ContainerAccess { + if base == nil && override == nil { + return nil + } + out := &ContainerAccess{} + merge := func(src *ContainerAccess) { + if src == nil { + return + } + out.Devices = appendUniqueStrings(out.Devices, src.Devices...) + out.PartitionRemoveEnv = appendUniqueStrings(out.PartitionRemoveEnv, src.PartitionRemoveEnv...) + out.Volumes = append(out.Volumes, src.Volumes...) + if out.Env == nil { + out.Env = map[string]string{} + } + for k, v := range src.Env { + out.Env[k] = v + } + if out.Ulimits == nil { + out.Ulimits = map[string]string{} + } + for k, v := range src.Ulimits { + out.Ulimits[k] = v + } + if src.Security != nil { + copied := *src.Security + copied.SupplementalGroups = append([]int(nil), src.Security.SupplementalGroups...) + out.Security = &copied + } + if src.DockerRuntime != "" { + out.DockerRuntime = src.DockerRuntime + } + if src.NetworkMode != "" { + out.NetworkMode = src.NetworkMode + } + if src.ShmSize != "" { + out.ShmSize = src.ShmSize + } + if src.Init { + out.Init = true + } + } + merge(base) + merge(override) + return out +} + +func appendUniqueStrings(values []string, additions ...string) []string { + seen := make(map[string]struct{}, len(values)+len(additions)) + for _, value := range values { + seen[value] = struct{}{} + } + for _, value := range additions { + if _, ok := seen[value]; ok { + continue + } + seen[value] = struct{}{} + values = append(values, value) + } + return values +} + // findRuntimeClassName looks up the K8s runtimeClassName from hardware profiles. // Returns "" if not specified (no runtimeClassName in pod spec). func (c *Catalog) findRuntimeClassName(hw HardwareInfo) string { diff --git a/internal/mcp/tools_deps.go b/internal/mcp/tools_deps.go index 03f0ea6c..b0dca101 100644 --- a/internal/mcp/tools_deps.go +++ b/internal/mcp/tools_deps.go @@ -171,7 +171,7 @@ type ToolDeps struct { // Scenario ScenarioList func(ctx context.Context) (json.RawMessage, error) ScenarioShow func(ctx context.Context, name string) (json.RawMessage, error) - ScenarioApply func(ctx context.Context, name string, dryRun bool) (json.RawMessage, error) + ScenarioApply func(ctx context.Context, name string, dryRun bool, bindings map[string]string) (json.RawMessage, error) // Explorer ExplorerStatus func(ctx context.Context) (json.RawMessage, error) diff --git a/internal/mcp/tools_scenario.go b/internal/mcp/tools_scenario.go index a66890f6..913fa4e4 100644 --- a/internal/mcp/tools_scenario.go +++ b/internal/mcp/tools_scenario.go @@ -38,18 +38,20 @@ func registerScenarioTools(s *Server, deps *ToolDeps) { // scenario.apply s.RegisterTool(&Tool{ Name: "scenario.apply", - Description: "Deploy all models defined in a deployment scenario. Supports dry_run to preview without executing.", + Description: "Deploy all models defined in a deployment scenario, including ordered deployments on remote Fleet devices. Supports dry_run to preview without executing.", InputSchema: schema( `"name":{"type":"string","description":"Scenario name, e.g. 'openclaw-multi'. Call catalog.list with kind=scenarios to see available scenarios."},`+ - `"dry_run":{"type":"boolean","description":"If true, preview deployment plans without executing (default false)"}`, + `"dry_run":{"type":"boolean","description":"If true, preview deployment plans without executing (default false)"},`+ + `"bindings":{"type":"object","additionalProperties":{"type":"string"},"description":"Values for scenario inputs such as Fleet device IDs and fabric addresses. Call scenario.show to inspect required inputs."}`, "name"), Handler: func(ctx context.Context, params json.RawMessage) (*ToolResult, error) { if deps.ScenarioApply == nil { return ErrorResult("scenario.apply not available"), nil } var p struct { - Name string `json:"name"` - DryRun bool `json:"dry_run"` + Name string `json:"name"` + DryRun bool `json:"dry_run"` + Bindings map[string]string `json:"bindings"` } if err := json.Unmarshal(params, &p); err != nil { return nil, fmt.Errorf("parse params: %w", err) @@ -57,7 +59,7 @@ func registerScenarioTools(s *Server, deps *ToolDeps) { if p.Name == "" { return ErrorResult("name is required"), nil } - data, err := deps.ScenarioApply(ctx, p.Name, p.DryRun) + data, err := deps.ScenarioApply(ctx, p.Name, p.DryRun, p.Bindings) if err != nil { return nil, fmt.Errorf("scenario apply: %w", err) } diff --git a/internal/runtime/docker.go b/internal/runtime/docker.go index e076a831..6aa7bc17 100644 --- a/internal/runtime/docker.go +++ b/internal/runtime/docker.go @@ -9,6 +9,7 @@ import ( "os" "os/exec" "path/filepath" + "sort" "strconv" "strings" "time" @@ -110,6 +111,16 @@ func (r *DockerRuntime) buildRunArgs(name string, req *DeployRequest) []string { if req.Container != nil && req.Container.ShmSize != "" { args = append(args, "--shm-size", req.Container.ShmSize) } + if req.Container != nil { + keys := make([]string, 0, len(req.Container.Ulimits)) + for key := range req.Container.Ulimits { + keys = append(keys, key) + } + sort.Strings(keys) + for _, key := range keys { + args = append(args, "--ulimit", key+"="+req.Container.Ulimits[key]) + } + } // Port publish (skip when using host network) if port := primaryPortForRequest(req); port > 0 && (req.Container == nil || req.Container.NetworkMode != "host") { diff --git a/internal/runtime/docker_test.go b/internal/runtime/docker_test.go index 9e225bb7..76efec27 100644 --- a/internal/runtime/docker_test.go +++ b/internal/runtime/docker_test.go @@ -302,6 +302,7 @@ func TestBuildRunArgs_Ascend(t *testing.T) { DockerRuntime: "ascend", NetworkMode: "host", ShmSize: "500g", + Ulimits: map[string]string{"memlock": "-1:-1"}, Init: true, Devices: []string{"/dev/davinci0", "/dev/davinci_manager", "/dev/devmm_svm", "/dev/hisi_hdc"}, Env: map[string]string{"PYTORCH_NPU_ALLOC_CONF": "expandable_segments:True"}, @@ -317,6 +318,7 @@ func TestBuildRunArgs_Ascend(t *testing.T) { assertContains(t, argStr, "--init", "init flag") assertContains(t, argStr, "--network host", "host network") assertContains(t, argStr, "--shm-size 500g", "shared memory size") + assertContains(t, argStr, "--ulimit memlock=-1:-1", "memlock ulimit") assertContains(t, argStr, "--privileged", "privileged mode") assertContains(t, argStr, "--device /dev/davinci0", "davinci device") assertContains(t, argStr, "--device /dev/davinci_manager", "davinci manager device") From 7ead7bda36439c6c513da7308c921179e3066296 Mon Sep 17 00:00:00 2001 From: xu16601526267 <264125260+xu16601526267@users.noreply.github.com> Date: Thu, 6 Aug 2026 13:39:28 +0800 Subject: [PATCH 3/3] feat(catalog): add DeepSeek V4 Flash dual-Spark deployment --- catalog/engines/vllm-dspark-gb10.yaml | 117 +++++++++++++++ catalog/models/deepseek-v4-flash-0731.yaml | 55 ++++++++ .../deepseek-v4-flash-dspark-2node.yaml | 112 +++++++++++++++ cmd/aima/tooldeps_model.go | 31 ++-- cmd/aima/tooldeps_model_test.go | 12 ++ internal/knowledge/loader_test.go | 70 +++++++++ internal/sqlite.go | 41 +++--- internal/ui/handler_test.go | 4 + internal/ui/static/index.html | 133 +++++++++++++++++- 9 files changed, 535 insertions(+), 40 deletions(-) create mode 100644 catalog/engines/vllm-dspark-gb10.yaml create mode 100644 catalog/models/deepseek-v4-flash-0731.yaml create mode 100644 catalog/scenarios/deepseek-v4-flash-dspark-2node.yaml diff --git a/catalog/engines/vllm-dspark-gb10.yaml b/catalog/engines/vllm-dspark-gb10.yaml new file mode 100644 index 00000000..7f3a05ed --- /dev/null +++ b/catalog/engines/vllm-dspark-gb10.yaml @@ -0,0 +1,117 @@ +kind: engine_asset +_profile: vllm +metadata: + name: vllm-dspark-gb10 + type: vllm + version: "anemll-0.1.1" + supported_formats: [safetensors] + supported_model_types: [llm] +hardware: + gpu_arch: Blackwell + vram_min_mib: 4096 +image: + name: ghcr.io/anemll/dspark-vllm-gx10 + tag: "0.1.1" + platforms: [linux/arm64] + registries: [ghcr.io] +patterns: + - "dspark-vllm-gx10" +container: + devices: + - /dev/infiniband + network_mode: host + shm_size: 64gb + ulimits: + memlock: "-1:-1" + stack: "67108864:67108864" +startup: + init_commands: + - "ulimit -l unlimited; ulimit -s 65536" + - "if [ -f /models/encoding/encoding_dsv4.py ]; then cp -f /models/encoding/encoding_dsv4.py /usr/local/lib/python3.12/dist-packages/vllm/tokenizers/deepseek_v4_encoding.py; fi" + env: + HF_HOME: /cache/huggingface + HF_HUB_DISABLE_XET: "1" + VLLM_CACHE_ROOT: /cache/huggingface/vllm-cache + VLLM_ALLOW_LONG_MAX_MODEL_LEN: "1" + VLLM_SPARSE_INDEXER_MAX_LOGITS_MB: "256" + VLLM_MEMORY_PROFILER_ESTIMATE_CUDAGRAPHS: "0" + VLLM_USE_FLASHINFER_SAMPLER: "1" + VLLM_USE_B12X_MOE: "1" + VLLM_USE_B12X_WO_PROJECTION: "1" + VLLM_DSPARK_GPU_REJECTED_CONTEXT_MASK: "1" + VLLM_USE_BREAKABLE_CUDAGRAPH: "0" + TORCH_CUDA_ARCH_LIST: "12.1a" + FLASHINFER_CUDA_ARCH_LIST: "12.1a" + CUTE_DSL_ARCH: sm_121a + FLASHINFER_DISABLE_VERSION_CHECK: "1" + FLASHINFER_WORKSPACE_BASE: /cache/huggingface/flashinfer + PYTORCH_CUDA_ALLOC_CONF: expandable_segments:True + NCCL_NET: IB + NCCL_IB_DISABLE: "0" + NCCL_IB_ADDR_FAMILY: AF_INET + NCCL_IB_ROCE_VERSION_NUM: "2" + NCCL_CROSS_NIC: "1" + NCCL_CUMEM_ENABLE: "0" + NCCL_IGNORE_CPU_AFFINITY: "1" + NCCL_NVLS_ENABLE: "0" + NCCL_DEBUG: WARN + default_args: + port: 8888 + host: 0.0.0.0 + served_model_name: deepseek-v4-flash-dspark + trust_remote_code: true + tensor_parallel_size: 2 + pipeline_parallel_size: 1 + kv_cache_dtype: nvfp4_ds_mla + block_size: 256 + max_model_len: 1048576 + max_num_seqs: 6 + max_num_batched_tokens: 8192 + max_cudagraph_capture_size: 36 + gpu_memory_utilization: 0.80 + enable_prefix_caching: true + enable_prompt_tokens_details: true + async_scheduling: true + enable_chunked_prefill: true + speculative_config: + method: dspark + num_speculative_tokens: 5 + draft_sample_method: probabilistic + tokenizer_mode: deepseek_v4 + distributed_executor_backend: mp + moe_backend: flashinfer_b12x + tool_call_parser: deepseek_v4 + enable_auto_tool_choice: true + reasoning_parser: deepseek_v4 + reasoning_config: + reasoning_parser: deepseek_v4 + reasoning_start_str: + reasoning_end_str: + default_chat_template_kwargs: + thinking: false + generation_config: vllm + enable_flashinfer_autotune: true + nnodes: 2 + master_port: 25000 + health_check: + path: /health + timeout_s: 1200 + warmup: + enabled: true + prompt: "Return the word ready." + max_tokens: 8 + timeout_s: 180 +amplifier: + features: + - dspark_mtp + - nvfp4_ds_mla + - flashinfer_b12x +time_constraints: + cold_start_s: [600, 1200] + model_switch_s: [600, 1200] +power_constraints: + typical_draw_watts: [60, 100] +runtime: + default: container + platform_recommendations: + linux/arm64: container diff --git a/catalog/models/deepseek-v4-flash-0731.yaml b/catalog/models/deepseek-v4-flash-0731.yaml new file mode 100644 index 00000000..d7447f41 --- /dev/null +++ b/catalog/models/deepseek-v4-flash-0731.yaml @@ -0,0 +1,55 @@ +kind: model_asset +metadata: + name: deepseek-v4-flash-0731 + type: llm + family: deepseek + # 284B target model plus the attached DSpark speculative module. The + # ~45.61 GB shown by some quantized mirrors is artifact size, not parameter count. + parameter_count: "304B" + released_at: "2026-07-31" + aliases: + - DeepSeek-V4-Flash-0731 + - DeepSeek-V4-Flash-DSpark +capabilities: + deployment_scenario: deepseek-v4-flash-dspark-2node +storage: + formats: [safetensors] + default_path_pattern: "{{.DataDir}}/models/{{.Name}}" + sources: + - type: huggingface + repo: deepseek-ai/DeepSeek-V4-Flash-0731 + format: safetensors + quantization: nvfp4 + - type: modelscope + repo: 0xSero/deepseek-v4-flash-0731-spark + format: safetensors + quantization: nvfp4 + - type: local_path + path: "" +variants: + - name: deepseek-v4-flash-0731-dspark-2node-nvfp4 + hardware: + gpu_arch: Blackwell + vram_min_mib: 0 + ram_min_mib: 98304 + # One GB10 is present on each Fleet device; the scenario supplies two nodes. + gpu_count_min: 1 + unified_memory: true + engine: vllm-dspark-gb10 + format: safetensors + source: + type: huggingface + repo: deepseek-ai/DeepSeek-V4-Flash-0731 + format: safetensors + quantization: nvfp4 + default_config: + max_model_len: 1048576 + max_num_seqs: 6 + max_num_batched_tokens: 8192 + gpu_memory_utilization: 0.80 + served_model_name: deepseek-v4-flash-dspark + expected_performance: + startup_time_s: 900 + cold_start_time_s: 1200 + vram_mib: 110000 + notes: "Two-node DSpark NVFP4 deployment. Exact throughput depends on RoCE wiring, GID selection, context length, and MTP warmup." diff --git a/catalog/scenarios/deepseek-v4-flash-dspark-2node.yaml b/catalog/scenarios/deepseek-v4-flash-dspark-2node.yaml new file mode 100644 index 00000000..3e2c2e49 --- /dev/null +++ b/catalog/scenarios/deepseek-v4-flash-dspark-2node.yaml @@ -0,0 +1,112 @@ +kind: deployment_scenario +metadata: + name: deepseek-v4-flash-dspark-2node + description: "DeepSeek V4 Flash DSpark across two DGX Spark GB10 devices using AIMA Fleet" +target: + hardware_profile: nvidia-gb10-arm64 + +inputs: + - name: worker_device + label: Worker Fleet device + description: "AIMA Fleet device ID for rank 1" + kind: device + required: true + - name: master_addr + label: Master fabric address + description: "Head-node IPv4 address reachable over the direct RoCE link" + kind: string + required: true + - name: head_host_ip + label: Head VLLM host IP + description: "Head-node address advertised by vLLM, normally the master fabric address" + kind: string + required: true + - name: worker_host_ip + label: Worker VLLM host IP + description: "Worker-node address on the direct RoCE link" + kind: string + required: true + - name: head_socket_ifname + label: Head fabric interface + description: "Head NCCL/Gloo interface name" + kind: string + required: true + - name: worker_socket_ifname + label: Worker fabric interface + description: "Worker NCCL/Gloo interface name" + kind: string + required: true + - name: head_ib_hca + label: Head IB HCA + description: "Head NCCL_IB_HCA value" + kind: string + required: true + - name: worker_ib_hca + label: Worker IB HCA + description: "Worker NCCL_IB_HCA value" + kind: string + required: true + - name: head_gid_index + label: Head RoCE GID index + description: "RoCEv2 GID index matching the head fabric address" + kind: string + required: true + - name: worker_gid_index + label: Worker RoCE GID index + description: "RoCEv2 GID index matching the worker fabric address" + kind: string + required: true + +deployments: + - id: worker + device: "{{.worker_device}}" + model: deepseek-v4-flash-0731 + engine: vllm-dspark-gb10 + no_pull: true + role: worker + config: + node_rank: 1 + headless: true + master_addr: "{{.master_addr}}" + env: + VLLM_HOST_IP: "{{.worker_host_ip}}" + NCCL_SOCKET_IFNAME: "{{.worker_socket_ifname}}" + TP_SOCKET_IFNAME: "{{.worker_socket_ifname}}" + GLOO_SOCKET_IFNAME: "{{.worker_socket_ifname}}" + NCCL_IB_HCA: "{{.worker_ib_hca}}" + NCCL_IB_GID_INDEX: "{{.worker_gid_index}}" + notes: "Rank 1 starts first and waits for the head rendezvous." + + - id: head + device: local + model: deepseek-v4-flash-0731 + engine: vllm-dspark-gb10 + no_pull: true + role: head + config: + node_rank: 0 + master_addr: "{{.master_addr}}" + env: + VLLM_HOST_IP: "{{.head_host_ip}}" + NCCL_SOCKET_IFNAME: "{{.head_socket_ifname}}" + TP_SOCKET_IFNAME: "{{.head_socket_ifname}}" + GLOO_SOCKET_IFNAME: "{{.head_socket_ifname}}" + NCCL_IB_HCA: "{{.head_ib_hca}}" + NCCL_IB_GID_INDEX: "{{.head_gid_index}}" + notes: "Rank 0 exposes the OpenAI-compatible API on port 8888." + +startup_order: + - step: 1 + deployment: worker + wait_for: "" + timeout_s: 0 + notes: "Launch the headless worker before rank 0." + - step: 2 + deployment: head + wait_for: health_check + timeout_s: 1200 + notes: "Wait for model load, graph capture, and the API health check." + +memory_budget: + total_unified_mib: 262144 + notes: "Two 128GB unified-memory nodes; per-node limits and KV-cache allocation are controlled by the model variant." diff --git a/cmd/aima/tooldeps_model.go b/cmd/aima/tooldeps_model.go index e537d07f..e922d30e 100644 --- a/cmd/aima/tooldeps_model.go +++ b/cmd/aima/tooldeps_model.go @@ -60,19 +60,20 @@ func registerCatalogLocalModel(ctx context.Context, ma *knowledge.ModelAsset, db continue } return db.UpsertScannedModel(ctx, &state.Model{ - ID: fmt.Sprintf("%x", sha256.Sum256([]byte(candidate.path+"|"+ma.Metadata.Name))), - Name: ma.Metadata.Name, - Type: ma.Metadata.Type, - Path: candidate.path, - Format: candidate.format, - SizeBytes: existingSizes[candidate.path], - DetectedArch: candidate.detectedArch, - ModelClass: strings.TrimSpace(ma.Metadata.ModelClass), - UIRole: strings.TrimSpace(ma.UI.Role), - UIDisplayNote: strings.TrimSpace(ma.UI.DisplayNote), - UIDisplayNoteZh: strings.TrimSpace(ma.UI.DisplayNoteZh), - StandaloneDeploy: ma.Capabilities.StandaloneDeploy, - Status: "registered", + ID: fmt.Sprintf("%x", sha256.Sum256([]byte(candidate.path+"|"+ma.Metadata.Name))), + Name: ma.Metadata.Name, + Type: ma.Metadata.Type, + Path: candidate.path, + Format: candidate.format, + SizeBytes: existingSizes[candidate.path], + DetectedArch: candidate.detectedArch, + ModelClass: strings.TrimSpace(ma.Metadata.ModelClass), + UIRole: strings.TrimSpace(ma.UI.Role), + UIDisplayNote: strings.TrimSpace(ma.UI.DisplayNote), + UIDisplayNoteZh: strings.TrimSpace(ma.UI.DisplayNoteZh), + StandaloneDeploy: ma.Capabilities.StandaloneDeploy, + DeploymentScenario: strings.TrimSpace(ma.Capabilities.DeploymentScenario), + Status: "registered", }) } return nil @@ -175,6 +176,9 @@ func annotateModelsFromCatalog(models []*state.Model, cat *knowledge.Catalog) { if m.StandaloneDeploy == nil { m.StandaloneDeploy = ma.Capabilities.StandaloneDeploy } + if strings.TrimSpace(m.DeploymentScenario) == "" { + m.DeploymentScenario = strings.TrimSpace(ma.Capabilities.DeploymentScenario) + } } // Speculative draft heads (e.g. DFlash/MTP) are companions of their @@ -315,6 +319,7 @@ func buildModelDeps(ac *appContext, deps *mcp.ToolDeps, if err != nil { return nil, err } + annotateModelsFromCatalog([]*state.Model{m}, cat) return json.Marshal(m) } diff --git a/cmd/aima/tooldeps_model_test.go b/cmd/aima/tooldeps_model_test.go index f61cc464..41e2ed97 100644 --- a/cmd/aima/tooldeps_model_test.go +++ b/cmd/aima/tooldeps_model_test.go @@ -29,6 +29,18 @@ func boolPtr(value bool) *bool { return &value } +func TestAnnotateModelsFromCatalogAddsDeploymentScenario(t *testing.T) { + models := []*state.Model{{Name: "deepseek-v4-flash-0731"}} + cat := &knowledge.Catalog{ModelAssets: []knowledge.ModelAsset{{ + Metadata: knowledge.ModelMetadata{Name: "deepseek-v4-flash-0731"}, + Capabilities: knowledge.ModelCapabilities{DeploymentScenario: "deepseek-v4-flash-dspark-2node"}, + }}} + annotateModelsFromCatalog(models, cat) + if got := models[0].DeploymentScenario; got != "deepseek-v4-flash-dspark-2node" { + t.Fatalf("deployment scenario = %q", got) + } +} + func TestScanModelsPublishesModelDiscoveredOnlyForNewModels(t *testing.T) { ctx := context.Background() db := mustOpenTooldepsDB(t) diff --git a/internal/knowledge/loader_test.go b/internal/knowledge/loader_test.go index 5b8eaef6..0e86c654 100644 --- a/internal/knowledge/loader_test.go +++ b/internal/knowledge/loader_test.go @@ -408,6 +408,76 @@ func TestScenarioNewFields(t *testing.T) { } } +func TestDeepSeekV4DSparkCatalogAssets(t *testing.T) { + cat, err := LoadCatalog(catalogFS()) + if err != nil { + t.Fatalf("load catalog: %v", err) + } + var model *ModelAsset + for i := range cat.ModelAssets { + if cat.ModelAssets[i].Metadata.Name == "deepseek-v4-flash-0731" { + model = &cat.ModelAssets[i] + break + } + } + if model == nil { + t.Fatal("deepseek-v4-flash-0731 model asset not found") + } + if model.Metadata.ParameterCount != "304B" || model.Capabilities.DeploymentScenario != "deepseek-v4-flash-dspark-2node" { + t.Fatalf("unexpected model metadata: %#v capabilities=%#v", model.Metadata, model.Capabilities) + } + var engine *EngineAsset + for i := range cat.EngineAssets { + if cat.EngineAssets[i].Metadata.Name == "vllm-dspark-gb10" { + engine = &cat.EngineAssets[i] + break + } + } + if engine == nil || engine.Container == nil { + t.Fatal("vllm-dspark-gb10 container metadata not found") + } + if engine.Container.NetworkMode != "host" || len(engine.Container.Devices) == 0 || engine.Container.Devices[0] != "/dev/infiniband" { + t.Fatalf("unexpected DSpark container access: %#v", engine.Container) + } + if got := engine.Startup.DefaultArgs["max_model_len"]; got != 1048576 { + t.Fatalf("max_model_len = %#v, want 1048576", got) + } + var scenario *DeploymentScenario + for i := range cat.DeploymentScenarios { + if cat.DeploymentScenarios[i].Metadata.Name == "deepseek-v4-flash-dspark-2node" { + scenario = &cat.DeploymentScenarios[i] + break + } + } + if scenario == nil || len(scenario.Inputs) == 0 || len(scenario.Deployments) != 2 { + t.Fatalf("unexpected DSpark scenario: %#v", scenario) + } + if scenario.Deployments[0].ID != "worker" || scenario.StartupOrder[0].Deployment != "worker" { + t.Fatalf("worker-first order not preserved: %#v", scenario.StartupOrder) + } + resolved, err := cat.Resolve(HardwareInfo{ + GPUArch: "Blackwell", GPUVRAMMiB: 15360, GPUCount: 1, UnifiedMemory: true, + CPUArch: "arm64", RAMTotalMiB: 131072, Platform: "linux/arm64", HardwareProfile: "nvidia-gb10-arm64", + }, model.Metadata.Name, engine.Metadata.Name, map[string]any{"model_path": "/models/deepseek-v4-flash-0731"}) + if err != nil { + t.Fatalf("resolve DSpark catalog assets: %v", err) + } + if resolved.Container == nil || resolved.Container.NetworkMode != "host" { + t.Fatalf("engine container access was not merged: %#v", resolved.Container) + } + podYAML, err := GeneratePod(resolved) + if err != nil { + t.Fatalf("generate DSpark pod: %v", err) + } + podText := string(podYAML) + if !strings.Contains(podText, `--max-model-len 1048576`) || strings.Contains(podText, "1.048576e+06") { + t.Fatalf("max_model_len was not rendered as a decimal integer:\n%s", podText) + } + if !strings.Contains(podText, "hostNetwork: true") || !strings.Contains(podText, "/dev/infiniband") { + t.Fatalf("DSpark pod is missing distributed container access:\n%s", podText) + } +} + func TestLoadCatalogInvalidYAML(t *testing.T) { fs := fstest.MapFS{ "hardware/bad.yaml": &fstest.MapFile{Data: []byte("not: valid: yaml: [")}, diff --git a/internal/sqlite.go b/internal/sqlite.go index b3e26438..53c13ad4 100644 --- a/internal/sqlite.go +++ b/internal/sqlite.go @@ -33,26 +33,27 @@ func (d *DB) RawDB() *sql.DB { } type Model struct { - ID string `json:"id"` - Name string `json:"name"` - Type string `json:"type"` - Path string `json:"path"` - Format string `json:"format"` - SizeBytes int64 `json:"size_bytes"` - DetectedArch string `json:"detected_arch"` - DetectedParams string `json:"detected_params"` - ModelClass string `json:"model_class"` - UIRole string `json:"ui_role"` - UIDisplayNote string `json:"ui_display_note"` - UIDisplayNoteZh string `json:"ui_display_note_zh"` - StandaloneDeploy *bool `json:"standalone_deploy,omitempty"` - TotalParams int64 `json:"total_params"` - ActiveParams int64 `json:"active_params"` - Quantization string `json:"quantization"` - QuantSrc string `json:"quant_src"` - Status string `json:"status"` - DownloadProgress float64 `json:"download_progress"` - CreatedAt time.Time `json:"created_at"` + ID string `json:"id"` + Name string `json:"name"` + Type string `json:"type"` + Path string `json:"path"` + Format string `json:"format"` + SizeBytes int64 `json:"size_bytes"` + DetectedArch string `json:"detected_arch"` + DetectedParams string `json:"detected_params"` + ModelClass string `json:"model_class"` + UIRole string `json:"ui_role"` + UIDisplayNote string `json:"ui_display_note"` + UIDisplayNoteZh string `json:"ui_display_note_zh"` + StandaloneDeploy *bool `json:"standalone_deploy,omitempty"` + DeploymentScenario string `json:"deployment_scenario,omitempty"` + TotalParams int64 `json:"total_params"` + ActiveParams int64 `json:"active_params"` + Quantization string `json:"quantization"` + QuantSrc string `json:"quant_src"` + Status string `json:"status"` + DownloadProgress float64 `json:"download_progress"` + CreatedAt time.Time `json:"created_at"` } func boolPtrToNullBool(value *bool) sql.NullBool { diff --git a/internal/ui/handler_test.go b/internal/ui/handler_test.go index f6fbbb48..935ef347 100644 --- a/internal/ui/handler_test.go +++ b/internal/ui/handler_test.go @@ -550,6 +550,10 @@ func TestRegisterRoutes_IndexDeployDetailUsesBackendDefaultsAndImmediateClose(t `typeof c.image_available_in_containerd === 'boolean'`, `.deploy-compat-grid,`, `model.detected_arch`, + `this.callTool('scenario.apply', { name: this.deployScenarioName, bindings: scenarioBindings })`, + `this.callTool('scenario.apply', {`, + `this.callTool('scenario.show', { name: scenarioName })`, + `x-text="t('deploy_cluster_config')"`, } { if !strings.Contains(body, token) { t.Fatalf("body missing deploy detail token %q", token) diff --git a/internal/ui/static/index.html b/internal/ui/static/index.html index 7e321cb1..3b25b92a 100644 --- a/internal/ui/static/index.html +++ b/internal/ui/static/index.html @@ -5962,6 +5962,30 @@

+
+
+
+ +
+
+
+
@@ -5992,7 +6016,7 @@

-
+
@@ -6023,7 +6047,7 @@

-
+