454 lines
15 KiB
Go
454 lines
15 KiB
Go
package node
|
|
|
|
import (
|
|
"fmt"
|
|
"strings"
|
|
|
|
"google.golang.org/protobuf/types/known/structpb"
|
|
|
|
"iop/packages/go/config"
|
|
iop "iop/proto/gen/iop"
|
|
)
|
|
|
|
// BuildConfigPayload converts a NodeRecord's adapter config to the proto payload
|
|
// sent to node during registration.
|
|
//
|
|
// Provider-first entries in rec.Providers are compiled first: each provider id
|
|
// becomes the adapter instance key. Legacy adapter instances in rec.Adapters are
|
|
// appended afterwards. An instance key that appears in both produces an error.
|
|
// Providers with a non-empty Adapter field and no protocol profile reference a
|
|
// legacy instance and are skipped during provider-first compile. A profile-
|
|
// backed provider is compiled under its provider id from the referenced HTTP
|
|
// adapter while the legacy instance is retained below.
|
|
//
|
|
// After config normalisation, OllamaInstances, VllmInstances, and
|
|
// OpenAICompatInstances already contain the promoted legacy single-instance
|
|
// entries, so only the slice fields are iterated here. Duplicate instance names
|
|
// produce an error.
|
|
//
|
|
// Mock adapter is only included when rec.Adapters.Mock.Enabled is true.
|
|
func BuildConfigPayload(rec *NodeRecord) (*iop.NodeConfigPayload, error) {
|
|
payload := &iop.NodeConfigPayload{
|
|
Runtime: &iop.NodeRuntimeConfig{
|
|
Concurrency: int32(rec.Runtime.Concurrency),
|
|
},
|
|
}
|
|
|
|
// Conditional mock adapter: only when explicitly enabled in config.
|
|
if rec.Adapters.Mock.Enabled {
|
|
payload.Adapters = append(payload.Adapters, &iop.AdapterConfig{
|
|
Type: "mock",
|
|
Enabled: true,
|
|
Config: &iop.AdapterConfig_Mock{Mock: &iop.MockAdapterConfig{}},
|
|
})
|
|
}
|
|
|
|
// flat namespace to detect conflicts between provider ids and legacy adapter keys
|
|
seenFlat := make(map[string]struct{})
|
|
|
|
for _, p := range rec.Providers {
|
|
if p.Adapter != "" && p.RuntimeProfile == nil {
|
|
// references a legacy adapter instance; skip provider-first compile
|
|
continue
|
|
}
|
|
if rec.Adapters.Mock.Enabled && p.ID == "mock" {
|
|
return nil, fmt.Errorf("provider id %q conflicts with explicit mock adapter instance key", p.ID)
|
|
}
|
|
if _, dup := seenFlat[p.ID]; dup {
|
|
return nil, fmt.Errorf("duplicate provider id %q", p.ID)
|
|
}
|
|
seenFlat[p.ID] = struct{}{}
|
|
var ac *iop.AdapterConfig
|
|
var err error
|
|
if p.RuntimeProfile != nil && strings.TrimSpace(p.Adapter) != "" {
|
|
ac, err = profileBackedProviderToAdapterConfig(rec, p)
|
|
} else {
|
|
ac, err = providerToAdapterConfig(p)
|
|
}
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
payload.Adapters = append(payload.Adapters, ac)
|
|
}
|
|
|
|
seen := make(map[string]struct{})
|
|
|
|
for _, inst := range rec.Adapters.OllamaInstances {
|
|
if !inst.Enabled {
|
|
continue
|
|
}
|
|
key := "ollama:" + inst.Name
|
|
if _, dup := seen[key]; dup {
|
|
return nil, fmt.Errorf("duplicate ollama instance name %q", inst.Name)
|
|
}
|
|
seen[key] = struct{}{}
|
|
if _, conflict := seenFlat[inst.Name]; conflict {
|
|
return nil, fmt.Errorf("ollama adapter instance key %q conflicts with provider id", inst.Name)
|
|
}
|
|
payload.Adapters = append(payload.Adapters, &iop.AdapterConfig{
|
|
Type: "ollama",
|
|
Enabled: true,
|
|
Name: inst.Name,
|
|
Config: &iop.AdapterConfig_Ollama{
|
|
Ollama: &iop.OllamaAdapterConfig{
|
|
BaseUrl: inst.BaseURL,
|
|
ContextSize: int32(inst.ContextSize),
|
|
Capacity: int32(inst.Capacity),
|
|
MaxQueue: int32(inst.MaxQueue),
|
|
QueueTimeoutMs: int32(inst.QueueTimeoutMS),
|
|
RequestTimeoutMs: int32(inst.RequestTimeoutMS),
|
|
},
|
|
},
|
|
})
|
|
}
|
|
|
|
for _, inst := range rec.Adapters.VllmInstances {
|
|
if !inst.Enabled {
|
|
continue
|
|
}
|
|
key := "vllm:" + inst.Name
|
|
if _, dup := seen[key]; dup {
|
|
return nil, fmt.Errorf("duplicate vllm instance name %q", inst.Name)
|
|
}
|
|
seen[key] = struct{}{}
|
|
if _, conflict := seenFlat[inst.Name]; conflict {
|
|
return nil, fmt.Errorf("vllm adapter instance key %q conflicts with provider id", inst.Name)
|
|
}
|
|
payload.Adapters = append(payload.Adapters, &iop.AdapterConfig{
|
|
Type: "vllm",
|
|
Enabled: true,
|
|
Name: inst.Name,
|
|
Config: &iop.AdapterConfig_Vllm{
|
|
Vllm: &iop.VllmAdapterConfig{
|
|
Endpoint: inst.Endpoint,
|
|
Capacity: int32(inst.Capacity),
|
|
MaxQueue: int32(inst.MaxQueue),
|
|
QueueTimeoutMs: int32(inst.QueueTimeoutMS),
|
|
RequestTimeoutMs: int32(inst.RequestTimeoutMS),
|
|
},
|
|
},
|
|
})
|
|
}
|
|
|
|
for _, inst := range rec.Adapters.OpenAICompatInstances {
|
|
if !inst.Enabled {
|
|
continue
|
|
}
|
|
key := "openai_compat:" + inst.Name
|
|
if _, dup := seen[key]; dup {
|
|
return nil, fmt.Errorf("duplicate openai_compat instance name %q", inst.Name)
|
|
}
|
|
seen[key] = struct{}{}
|
|
if _, conflict := seenFlat[inst.Name]; conflict {
|
|
return nil, fmt.Errorf("openai_compat adapter instance key %q conflicts with provider id", inst.Name)
|
|
}
|
|
payload.Adapters = append(payload.Adapters, &iop.AdapterConfig{
|
|
Type: "openai_compat",
|
|
Enabled: true,
|
|
Name: inst.Name,
|
|
Config: &iop.AdapterConfig_OpenaiCompat{
|
|
OpenaiCompat: &iop.OpenAICompatAdapterConfig{
|
|
Provider: inst.Provider,
|
|
Endpoint: inst.Endpoint,
|
|
Headers: inst.Headers,
|
|
Capacity: int32(inst.Capacity),
|
|
MaxQueue: int32(inst.MaxQueue),
|
|
QueueTimeoutMs: int32(inst.QueueTimeoutMS),
|
|
RequestTimeoutMs: int32(inst.RequestTimeoutMS),
|
|
},
|
|
},
|
|
})
|
|
}
|
|
|
|
if rec.Adapters.CLI.Enabled {
|
|
if _, conflict := seenFlat["cli"]; conflict {
|
|
return nil, fmt.Errorf("cli adapter instance key %q conflicts with provider id", "cli")
|
|
}
|
|
profiles := make(map[string]*iop.CLIProfileConfig, len(rec.Adapters.CLI.Profiles))
|
|
for name, p := range rec.Adapters.CLI.Profiles {
|
|
profiles[name] = cliProfileToProto(p)
|
|
}
|
|
payload.Adapters = append(payload.Adapters, &iop.AdapterConfig{
|
|
Type: "cli",
|
|
Enabled: true,
|
|
Config: &iop.AdapterConfig_Cli{
|
|
Cli: &iop.CLIAdapterConfig{Profiles: profiles},
|
|
},
|
|
})
|
|
}
|
|
return payload, nil
|
|
}
|
|
|
|
type profileBackingAdapter struct {
|
|
provider string
|
|
endpoint string
|
|
headers map[string]string
|
|
capacity int
|
|
maxQueue int
|
|
queueTimeoutMS int
|
|
requestTimeoutMS int
|
|
}
|
|
|
|
func resolveProfileBackingAdapter(rec *NodeRecord, providerID, ref string) (profileBackingAdapter, error) {
|
|
ref = strings.TrimSpace(ref)
|
|
type match struct {
|
|
enabled bool
|
|
backing profileBackingAdapter
|
|
typ string
|
|
}
|
|
var exact []match
|
|
for _, inst := range rec.Adapters.VllmInstances {
|
|
if inst.Name == ref {
|
|
exact = append(exact, match{inst.Enabled, profileBackingAdapter{
|
|
provider: "vllm", endpoint: inst.Endpoint, capacity: inst.Capacity,
|
|
maxQueue: inst.MaxQueue, queueTimeoutMS: inst.QueueTimeoutMS, requestTimeoutMS: inst.RequestTimeoutMS,
|
|
}, "vllm"})
|
|
}
|
|
}
|
|
for _, inst := range rec.Adapters.OpenAICompatInstances {
|
|
if inst.Name == ref {
|
|
exact = append(exact, match{inst.Enabled, profileBackingAdapter{
|
|
provider: inst.Provider, endpoint: inst.Endpoint, headers: cloneHeaders(inst.Headers), capacity: inst.Capacity,
|
|
maxQueue: inst.MaxQueue, queueTimeoutMS: inst.QueueTimeoutMS, requestTimeoutMS: inst.RequestTimeoutMS,
|
|
}, "openai_compat"})
|
|
}
|
|
}
|
|
if len(exact) > 1 {
|
|
return profileBackingAdapter{}, fmt.Errorf("provider %q adapter %q: backing adapter reference is ambiguous", providerID, ref)
|
|
}
|
|
if len(exact) == 1 {
|
|
if !exact[0].enabled {
|
|
return profileBackingAdapter{}, fmt.Errorf("provider %q adapter %q: backing %s adapter is disabled", providerID, ref, exact[0].typ)
|
|
}
|
|
return exact[0].backing, nil
|
|
}
|
|
|
|
var enabled []profileBackingAdapter
|
|
disabled := 0
|
|
switch ref {
|
|
case "vllm":
|
|
for _, inst := range rec.Adapters.VllmInstances {
|
|
if !inst.Enabled {
|
|
disabled++
|
|
continue
|
|
}
|
|
enabled = append(enabled, profileBackingAdapter{
|
|
provider: "vllm", endpoint: inst.Endpoint, capacity: inst.Capacity,
|
|
maxQueue: inst.MaxQueue, queueTimeoutMS: inst.QueueTimeoutMS, requestTimeoutMS: inst.RequestTimeoutMS,
|
|
})
|
|
}
|
|
case "openai_compat":
|
|
for _, inst := range rec.Adapters.OpenAICompatInstances {
|
|
if !inst.Enabled {
|
|
disabled++
|
|
continue
|
|
}
|
|
enabled = append(enabled, profileBackingAdapter{
|
|
provider: inst.Provider, endpoint: inst.Endpoint, headers: cloneHeaders(inst.Headers), capacity: inst.Capacity,
|
|
maxQueue: inst.MaxQueue, queueTimeoutMS: inst.QueueTimeoutMS, requestTimeoutMS: inst.RequestTimeoutMS,
|
|
})
|
|
}
|
|
default:
|
|
return profileBackingAdapter{}, fmt.Errorf("provider %q adapter %q: backing adapter is missing or unsupported", providerID, ref)
|
|
}
|
|
switch len(enabled) {
|
|
case 1:
|
|
return enabled[0], nil
|
|
case 0:
|
|
if disabled > 0 {
|
|
return profileBackingAdapter{}, fmt.Errorf("provider %q adapter %q: backing adapter has no enabled instance", providerID, ref)
|
|
}
|
|
return profileBackingAdapter{}, fmt.Errorf("provider %q adapter %q: backing adapter is missing", providerID, ref)
|
|
default:
|
|
return profileBackingAdapter{}, fmt.Errorf("provider %q adapter %q: backing adapter is ambiguous (%d enabled instances)", providerID, ref, len(enabled))
|
|
}
|
|
}
|
|
|
|
func profileBackedProviderToAdapterConfig(rec *NodeRecord, p config.NodeProviderConf) (*iop.AdapterConfig, error) {
|
|
backing, err := resolveProfileBackingAdapter(rec, p.ID, p.Adapter)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
provider := backing.provider
|
|
if overlay := openAICompatProviderLabel(p); overlay != "" {
|
|
provider = overlay
|
|
}
|
|
endpoint := backing.endpoint
|
|
if overlay := strings.TrimSpace(p.Endpoint); overlay != "" {
|
|
endpoint = overlay
|
|
} else if overlay := strings.TrimSpace(p.BaseURL); overlay != "" {
|
|
endpoint = overlay
|
|
}
|
|
headers := cloneHeaders(backing.headers)
|
|
if p.Headers != nil {
|
|
headers = cloneHeaders(p.Headers)
|
|
}
|
|
capacity := nonZeroOr(p.Capacity, backing.capacity)
|
|
maxQueue := nonZeroOr(p.MaxQueue, backing.maxQueue)
|
|
queueTimeoutMS := nonZeroOr(p.QueueTimeoutMS, backing.queueTimeoutMS)
|
|
requestTimeoutMS := nonZeroOr(p.RequestTimeoutMS, backing.requestTimeoutMS)
|
|
return &iop.AdapterConfig{
|
|
Type: "openai_compat", Enabled: true, Name: p.ID,
|
|
Config: &iop.AdapterConfig_OpenaiCompat{OpenaiCompat: &iop.OpenAICompatAdapterConfig{
|
|
Provider: provider, Endpoint: endpoint, Headers: headers,
|
|
Capacity: int32(capacity), MaxQueue: int32(maxQueue), QueueTimeoutMs: int32(queueTimeoutMS),
|
|
RequestTimeoutMs: int32(requestTimeoutMS), ProtocolProfile: concreteProfileToProto(p.RuntimeProfile),
|
|
}},
|
|
}, nil
|
|
}
|
|
|
|
func nonZeroOr(value, fallback int) int {
|
|
if value != 0 {
|
|
return value
|
|
}
|
|
return fallback
|
|
}
|
|
|
|
func cloneHeaders(headers map[string]string) map[string]string {
|
|
if headers == nil {
|
|
return nil
|
|
}
|
|
cloned := make(map[string]string, len(headers))
|
|
for key, value := range headers {
|
|
cloned[key] = value
|
|
}
|
|
return cloned
|
|
}
|
|
|
|
// providerToAdapterConfig compiles a provider-first NodeProviderConf into an
|
|
// AdapterConfig using the provider id as the instance key.
|
|
func providerToAdapterConfig(p config.NodeProviderConf) (*iop.AdapterConfig, error) {
|
|
typeName := config.NormalizeProviderType(p.Type)
|
|
switch typeName {
|
|
case "ollama":
|
|
return &iop.AdapterConfig{
|
|
Type: "ollama",
|
|
Enabled: true,
|
|
Name: p.ID,
|
|
Config: &iop.AdapterConfig_Ollama{
|
|
Ollama: &iop.OllamaAdapterConfig{
|
|
BaseUrl: p.BaseURL,
|
|
ContextSize: int32(p.ContextSize),
|
|
Capacity: int32(p.Capacity),
|
|
MaxQueue: int32(p.MaxQueue),
|
|
QueueTimeoutMs: int32(p.QueueTimeoutMS),
|
|
RequestTimeoutMs: int32(p.RequestTimeoutMS),
|
|
},
|
|
},
|
|
}, nil
|
|
|
|
case "openai_compat":
|
|
return &iop.AdapterConfig{
|
|
Type: "openai_compat",
|
|
Enabled: true,
|
|
Name: p.ID,
|
|
Config: &iop.AdapterConfig_OpenaiCompat{
|
|
OpenaiCompat: &iop.OpenAICompatAdapterConfig{
|
|
Provider: openAICompatProviderLabel(p),
|
|
Endpoint: p.Endpoint,
|
|
Headers: p.Headers,
|
|
Capacity: int32(p.Capacity),
|
|
MaxQueue: int32(p.MaxQueue),
|
|
QueueTimeoutMs: int32(p.QueueTimeoutMS),
|
|
RequestTimeoutMs: int32(p.RequestTimeoutMS),
|
|
ProtocolProfile: concreteProfileToProto(p.RuntimeProfile),
|
|
},
|
|
},
|
|
}, nil
|
|
|
|
case "cli":
|
|
// provider id becomes both the adapter instance key and the profile key
|
|
return &iop.AdapterConfig{
|
|
Type: "cli",
|
|
Enabled: true,
|
|
Name: p.ID,
|
|
Config: &iop.AdapterConfig_Cli{
|
|
Cli: &iop.CLIAdapterConfig{
|
|
Profiles: map[string]*iop.CLIProfileConfig{
|
|
p.ID: {
|
|
Command: p.Command,
|
|
Args: append([]string(nil), p.Args...),
|
|
Env: append([]string(nil), p.Env...),
|
|
Mode: p.Mode,
|
|
OutputFormat: p.OutputFormat,
|
|
ResumeArgs: append([]string(nil), p.ResumeArgs...),
|
|
},
|
|
},
|
|
},
|
|
},
|
|
}, nil
|
|
|
|
default:
|
|
return nil, fmt.Errorf("provider %q: unsupported type %q for provider-first compile", p.ID, p.Type)
|
|
}
|
|
}
|
|
|
|
func openAICompatProviderLabel(p config.NodeProviderConf) string {
|
|
if provider := strings.TrimSpace(p.Provider); provider != "" {
|
|
return provider
|
|
}
|
|
providerType := strings.ToLower(strings.TrimSpace(p.Type))
|
|
switch providerType {
|
|
case "vllm", "vllm-mlx", "lemonade", "sglang", "openai_api", "seulgivibe_claude", "seulgivibe_openai":
|
|
return providerType
|
|
default:
|
|
return ""
|
|
}
|
|
}
|
|
|
|
// concreteProfileToProto converts a resolved config.ConcreteProtocolProfile
|
|
// snapshot to its wire representation. Returns nil when the profile is absent
|
|
// so legacy payloads without a profile remain source-compatible.
|
|
func concreteProfileToProto(p *config.ConcreteProtocolProfile) *iop.ConcreteProtocolProfile {
|
|
if p == nil {
|
|
return nil
|
|
}
|
|
cp := p.Clone()
|
|
proto := &iop.ConcreteProtocolProfile{
|
|
Id: cp.ID,
|
|
Driver: string(cp.Driver),
|
|
BaseUrl: cp.BaseURL,
|
|
Auth: &iop.ProtocolAuth{Header: cp.Auth.Header, Scheme: cp.Auth.Scheme},
|
|
Capabilities: append([]string(nil), cp.Capabilities...),
|
|
}
|
|
if len(cp.Operations) > 0 {
|
|
proto.Operations = make(map[string]string, len(cp.Operations))
|
|
for k, v := range cp.Operations {
|
|
proto.Operations[k] = v
|
|
}
|
|
}
|
|
if len(cp.ModelMapping) > 0 {
|
|
proto.ModelMapping = make(map[string]string, len(cp.ModelMapping))
|
|
for k, v := range cp.ModelMapping {
|
|
proto.ModelMapping[k] = v
|
|
}
|
|
}
|
|
if len(cp.Extensions) > 0 {
|
|
if s, err := structpb.NewStruct(cp.Extensions); err == nil {
|
|
proto.Extensions = s
|
|
}
|
|
}
|
|
return proto
|
|
}
|
|
|
|
func cliProfileToProto(p config.CLIProfileConf) *iop.CLIProfileConfig {
|
|
out := &iop.CLIProfileConfig{
|
|
Command: p.Command,
|
|
Args: append([]string(nil), p.Args...),
|
|
Env: append([]string(nil), p.Env...),
|
|
Persistent: p.Persistent,
|
|
Terminal: p.Terminal,
|
|
ResponseIdleTimeoutMs: int32(p.ResponseIdleTimeoutMS),
|
|
StartupIdleTimeoutMs: int32(p.StartupIdleTimeoutMS),
|
|
OutputFormat: p.OutputFormat,
|
|
Mode: p.Mode,
|
|
ResumeArgs: append([]string(nil), p.ResumeArgs...),
|
|
}
|
|
if !p.CompletionMarker.Empty() {
|
|
out.CompletionMarker = &iop.CLICompletionMarker{
|
|
Line: p.CompletionMarker.Line,
|
|
Regex: p.CompletionMarker.Regex,
|
|
}
|
|
}
|
|
return out
|
|
}
|