iop/apps/edge/internal/configrefresh/classify.go
toki 367e7f3dc4 feat(openai): 모델 생성 정책을 적용한다
provider pool 모델 catalog의 출력 토큰 기본값과 thinking budget을 OpenAI-compatible 요청에 반영하고, Ornith dev-runtime 기준과 관련 계약 문서를 함께 맞춘다.
2026-07-04 09:20:56 +09:00

529 lines
19 KiB
Go
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

package configrefresh
import (
"context"
"fmt"
"os"
"path/filepath"
"reflect"
"sort"
"strings"
"iop/apps/edge/internal/edgevalidate"
"iop/packages/go/config"
)
// applyRuntimeNormalization normalizes candidate config so that classification
// uses the same effective values as a running serve process.
func applyRuntimeNormalization(cfg *config.EdgeConfig) {
// Normalize logging.path must match edgecmd.ResolveEdgeLogPath exactly:
// - empty → resolve to binary-dir/logs/edge.log
// - absolute → preserve exactly
// - relative → preserve exactly (include ./ segments)
// Serve keeps explicit relative paths like "./logs/edge.log" unchanged,
// so classifying against a cleaned value produces false restart_required.
if cfg.Logging.Path == "" {
cfg.Logging.Path = defaultEdgeLogPath()
}
// NOTE: explicit absolute / relative paths are intentionally left untouched.
// Normalize bootstrap.artifact_dir resolve relative paths to absolute.
cfg.Bootstrap.ArtifactDir = resolveBootstrapArtifactDir(cfg.Bootstrap.ArtifactDir)
}
func defaultEdgeLogPath() string {
if exe, err := os.Executable(); err == nil {
return filepath.Join(filepath.Dir(exe), "logs", "edge.log")
}
if wd, err := os.Getwd(); err == nil {
return filepath.Join(wd, "logs", "edge.log")
}
return "logs/edge.log"
}
func resolveBootstrapArtifactDir(dir string) string {
if dir == "" {
dir = "artifacts"
}
if filepath.IsAbs(dir) {
return dir
}
return filepath.Join(binaryDir(), dir)
}
func binaryDir() string {
if exe, err := os.Executable(); err == nil {
return filepath.Dir(exe)
}
if wd, err := os.Getwd(); err == nil {
return wd
}
return "."
}
// LoadCandidate loads and validates an EdgeConfig from the given path, then
// applies the same runtime normalization that iop-edge serve uses. This
// ensures that classification compares effective (normalized) values and
// prevents false restart_required results when source-level YAML is
// identical but paths are relative or empty.
func LoadCandidate(path string) (*config.EdgeConfig, error) {
cfg, err := config.LoadEdge(path)
if err != nil {
return nil, err
}
if err := edgevalidate.ValidateEdgeConfig(cfg); err != nil {
return nil, err
}
applyRuntimeNormalization(cfg)
return cfg, nil
}
// providerKey identifies a provider entry for diffing.
type providerKey struct {
NodeKey string
Type string
Category config.Category
Adapter string
Models []string
Health string
Capacity int
Priority int
MaxQueue int
QueueTimeoutMS int
LifecycleCapabilities []string
// Enabled tracks the effective enabled state for live-apply detection.
// Not used for restart-required structural comparison.
Enabled bool
// Provider-First execution fields (G06) — all are restart-required on change
// because they alter what the Node adapter connects to or how it runs.
Provider string
Endpoint string
BaseURL string
Headers map[string]string
Command string
Args []string
Env []string
Mode string
ResumeArgs []string
OutputFormat string
ContextSize int
RequestTimeoutMS int
}
func buildProviderIndex(cfg *config.EdgeConfig) map[string]providerKey {
idx := make(map[string]providerKey)
for i, node := range cfg.Nodes {
nodeKey := nodeIdentity(node, i)
for _, p := range node.Providers {
idx[p.ID] = providerKey{
NodeKey: nodeKey,
Type: p.Type,
Category: p.Category,
Adapter: p.Adapter,
Models: append([]string(nil), p.Models...),
Health: p.Health,
Capacity: p.Capacity,
Priority: p.Priority,
MaxQueue: p.MaxQueue,
QueueTimeoutMS: p.QueueTimeoutMS,
LifecycleCapabilities: append([]string(nil), p.LifecycleCapabilities...),
Enabled: config.ProviderEnabled(p),
Provider: p.Provider,
Endpoint: p.Endpoint,
BaseURL: p.BaseURL,
Headers: cloneStringMap(p.Headers),
Command: p.Command,
Args: append([]string(nil), p.Args...),
Env: append([]string(nil), p.Env...),
Mode: p.Mode,
ResumeArgs: append([]string(nil), p.ResumeArgs...),
OutputFormat: p.OutputFormat,
ContextSize: p.ContextSize,
RequestTimeoutMS: p.RequestTimeoutMS,
}
}
}
return idx
}
type nodeKey struct {
Alias string
Token string
AgentKind string
Adapters config.AdaptersConf
Runtime config.RuntimeConf
}
func buildNodeIndex(cfg *config.EdgeConfig) map[string]nodeKey {
idx := make(map[string]nodeKey, len(cfg.Nodes))
for i, node := range cfg.Nodes {
idx[nodeIdentity(node, i)] = nodeKey{
Alias: node.Alias,
Token: node.Token,
AgentKind: node.AgentKind,
Adapters: node.Adapters,
Runtime: node.Runtime,
}
}
return idx
}
func nodeIdentity(node config.NodeDefinition, index int) string {
if node.ID != "" {
return node.ID
}
if node.Alias != "" {
return node.Alias
}
return fmt.Sprintf("#%d", index)
}
func buildModelIndex(cfg *config.EdgeConfig) map[string]config.ModelCatalogEntry {
idx := make(map[string]config.ModelCatalogEntry, len(cfg.Models))
for _, model := range cfg.Models {
idx[model.ID] = model
}
return idx
}
func appendIfChanged[T comparable](changes *[]Change, path string, class Status, previous, next T) {
if previous == next {
return
}
*changes = append(*changes, Change{
Path: path,
Class: class,
Previous: fmt.Sprint(previous),
Next: fmt.Sprint(next),
})
}
func appendDeepIfChanged(changes *[]Change, path string, class Status, previous, next interface{}) {
if reflect.DeepEqual(previous, next) {
return
}
*changes = append(*changes, Change{
Path: path,
Class: class,
Previous: fmt.Sprintf("%v", previous),
Next: fmt.Sprintf("%v", next),
})
}
// Classify computes the diff between current and candidate configs and returns
// the classified change list and an overall status.
func Classify(current, candidate *config.EdgeConfig) Result {
var changes []Change
// --- restart_required fields ---
appendIfChanged(&changes, "edge.id", StatusRestartRequired, current.Edge.ID, candidate.Edge.ID)
appendIfChanged(&changes, "edge.name", StatusRestartRequired, current.Edge.Name, candidate.Edge.Name)
appendIfChanged(&changes, "server.listen", StatusRestartRequired, current.Server.Listen, candidate.Server.Listen)
appendIfChanged(&changes, "server.advertise_host", StatusRestartRequired, current.Server.AdvertiseHost, candidate.Server.AdvertiseHost)
appendIfChanged(&changes, "bootstrap.listen", StatusRestartRequired, current.Bootstrap.Listen, candidate.Bootstrap.Listen)
appendIfChanged(&changes, "bootstrap.artifact_base_url", StatusRestartRequired, current.Bootstrap.ArtifactBaseURL, candidate.Bootstrap.ArtifactBaseURL)
appendIfChanged(&changes, "bootstrap.artifact_dir", StatusRestartRequired, current.Bootstrap.ArtifactDir, candidate.Bootstrap.ArtifactDir)
appendDeepIfChanged(&changes, "tls", StatusRestartRequired, current.TLS, candidate.TLS)
appendDeepIfChanged(&changes, "logging", StatusRestartRequired, current.Logging, candidate.Logging)
appendIfChanged(&changes, "metrics.port", StatusRestartRequired, current.Metrics.Port, candidate.Metrics.Port)
appendDeepIfChanged(&changes, "console", StatusRestartRequired, current.Console, candidate.Console)
appendDeepIfChanged(&changes, "control_plane", StatusRestartRequired, current.ControlPlane, candidate.ControlPlane)
appendDeepIfChanged(&changes, "refresh", StatusRestartRequired, current.Refresh, candidate.Refresh)
appendDeepIfChanged(&changes, "openai", StatusRestartRequired, current.OpenAI, candidate.OpenAI)
appendDeepIfChanged(&changes, "a2a", StatusRestartRequired, current.A2A, candidate.A2A)
currentNodes := buildNodeIndex(current)
candidateNodes := buildNodeIndex(candidate)
for key, cur := range currentNodes {
next, exists := candidateNodes[key]
if !exists {
changes = append(changes, Change{
Path: fmt.Sprintf("nodes[%q]", key),
Class: StatusRestartRequired,
Previous: "present",
Next: "absent",
})
continue
}
appendIfChanged(&changes, fmt.Sprintf("nodes[%q].alias", key), StatusRestartRequired, cur.Alias, next.Alias)
appendIfChanged(&changes, fmt.Sprintf("nodes[%q].token", key), StatusRestartRequired, cur.Token, next.Token)
appendIfChanged(&changes, fmt.Sprintf("nodes[%q].agent_kind", key), StatusRestartRequired, cur.AgentKind, next.AgentKind)
appendDeepIfChanged(&changes, fmt.Sprintf("nodes[%q].adapters", key), StatusRestartRequired, cur.Adapters, next.Adapters)
// Legacy runtime concurrency metadata is live-applyable for compat.
// Runtime admission is owned by provider/resource capacity.
appendIfChanged(&changes, fmt.Sprintf("nodes[%q].runtime.concurrency", key), StatusApplied, cur.Runtime.Concurrency, next.Runtime.Concurrency)
}
for key := range candidateNodes {
if _, exists := currentNodes[key]; !exists {
changes = append(changes, Change{
Path: fmt.Sprintf("nodes[%q]", key),
Class: StatusRestartRequired,
Previous: "absent",
Next: "present",
})
}
}
// --- mutable applied fields: provider capacity ---
currentProviders := buildProviderIndex(current)
candidateProviders := buildProviderIndex(candidate)
for provID, cur := range currentProviders {
next, exists := candidateProviders[provID]
if !exists {
changes = append(changes, Change{
Path: fmt.Sprintf("nodes[].providers[%q]", provID),
Class: StatusRestartRequired,
Previous: "present",
Next: "absent",
})
continue
}
appendIfChanged(&changes, fmt.Sprintf("nodes[].providers[%q].node", provID), StatusRestartRequired, cur.NodeKey, next.NodeKey)
appendIfChanged(&changes, fmt.Sprintf("nodes[].providers[%q].type", provID), StatusRestartRequired, cur.Type, next.Type)
appendIfChanged(&changes, fmt.Sprintf("nodes[].providers[%q].category", provID), StatusRestartRequired, cur.Category, next.Category)
appendIfChanged(&changes, fmt.Sprintf("nodes[].providers[%q].adapter", provID), StatusRestartRequired, cur.Adapter, next.Adapter)
appendDeepIfChanged(&changes, fmt.Sprintf("nodes[].providers[%q].models", provID), StatusRestartRequired, cur.Models, next.Models)
appendIfChanged(&changes, fmt.Sprintf("nodes[].providers[%q].health", provID), StatusRestartRequired, cur.Health, next.Health)
appendDeepIfChanged(&changes, fmt.Sprintf("nodes[].providers[%q].lifecycle_capabilities", provID), StatusRestartRequired, cur.LifecycleCapabilities, next.LifecycleCapabilities)
appendIfChanged(&changes, fmt.Sprintf("nodes[].providers[%q].provider", provID), StatusRestartRequired, cur.Provider, next.Provider)
appendIfChanged(&changes, fmt.Sprintf("nodes[].providers[%q].endpoint", provID), StatusRestartRequired, cur.Endpoint, next.Endpoint)
appendIfChanged(&changes, fmt.Sprintf("nodes[].providers[%q].base_url", provID), StatusRestartRequired, cur.BaseURL, next.BaseURL)
appendDeepIfChanged(&changes, fmt.Sprintf("nodes[].providers[%q].headers", provID), StatusRestartRequired, cur.Headers, next.Headers)
appendIfChanged(&changes, fmt.Sprintf("nodes[].providers[%q].command", provID), StatusRestartRequired, cur.Command, next.Command)
appendDeepIfChanged(&changes, fmt.Sprintf("nodes[].providers[%q].args", provID), StatusRestartRequired, cur.Args, next.Args)
appendDeepIfChanged(&changes, fmt.Sprintf("nodes[].providers[%q].env", provID), StatusRestartRequired, cur.Env, next.Env)
appendIfChanged(&changes, fmt.Sprintf("nodes[].providers[%q].mode", provID), StatusRestartRequired, cur.Mode, next.Mode)
appendDeepIfChanged(&changes, fmt.Sprintf("nodes[].providers[%q].resume_args", provID), StatusRestartRequired, cur.ResumeArgs, next.ResumeArgs)
appendIfChanged(&changes, fmt.Sprintf("nodes[].providers[%q].output_format", provID), StatusRestartRequired, cur.OutputFormat, next.OutputFormat)
appendIfChanged(&changes, fmt.Sprintf("nodes[].providers[%q].context_size", provID), StatusRestartRequired, cur.ContextSize, next.ContextSize)
appendIfChanged(&changes, fmt.Sprintf("nodes[].providers[%q].request_timeout_ms", provID), StatusRestartRequired, cur.RequestTimeoutMS, next.RequestTimeoutMS)
}
for provID := range candidateProviders {
if _, exists := currentProviders[provID]; !exists {
changes = append(changes, Change{
Path: fmt.Sprintf("nodes[].providers[%q]", provID),
Class: StatusRestartRequired,
Previous: "absent",
Next: "present",
})
}
}
for provID, cp := range candidateProviders {
cur, exists := currentProviders[provID]
if !exists {
continue
}
if cur.Capacity != cp.Capacity {
changes = append(changes, Change{
Path: fmt.Sprintf("nodes[].providers[%q].capacity", provID),
Class: StatusApplied,
Previous: fmt.Sprintf("%d", cur.Capacity),
Next: fmt.Sprintf("%d", cp.Capacity),
})
}
if cur.Priority != cp.Priority {
changes = append(changes, Change{
Path: fmt.Sprintf("nodes[].providers[%q].priority", provID),
Class: StatusApplied,
Previous: fmt.Sprintf("%d", cur.Priority),
Next: fmt.Sprintf("%d", cp.Priority),
})
}
if cur.MaxQueue != cp.MaxQueue {
changes = append(changes, Change{
Path: fmt.Sprintf("nodes[].providers[%q].max_queue", provID),
Class: StatusApplied,
Previous: fmt.Sprintf("%d", cur.MaxQueue),
Next: fmt.Sprintf("%d", cp.MaxQueue),
})
}
if cur.QueueTimeoutMS != cp.QueueTimeoutMS {
changes = append(changes, Change{
Path: fmt.Sprintf("nodes[].providers[%q].queue_timeout_ms", provID),
Class: StatusApplied,
Previous: fmt.Sprintf("%d", cur.QueueTimeoutMS),
Next: fmt.Sprintf("%d", cp.QueueTimeoutMS),
})
}
// enabled is a live-apply switch: toggling it changes dispatch pool membership
// without requiring a process restart.
if cur.Enabled != cp.Enabled {
changes = append(changes, Change{
Path: fmt.Sprintf("nodes[].providers[%q].enabled", provID),
Class: StatusApplied,
Previous: fmt.Sprintf("%v", cur.Enabled),
Next: fmt.Sprintf("%v", cp.Enabled),
})
}
}
currentModels := buildModelIndex(current)
candidateModels := buildModelIndex(candidate)
for modelID, cur := range currentModels {
next, exists := candidateModels[modelID]
if !exists {
changes = append(changes, Change{
Path: fmt.Sprintf("models[%q]", modelID),
Class: StatusApplied,
Previous: "present",
Next: "absent",
})
continue
}
appendIfChanged(&changes, fmt.Sprintf("models[%q].display_name", modelID), StatusApplied, cur.DisplayName, next.DisplayName)
appendIfChanged(&changes, fmt.Sprintf("models[%q].default_max_tokens", modelID), StatusApplied, cur.DefaultMaxTokens, next.DefaultMaxTokens)
appendIfChanged(&changes, fmt.Sprintf("models[%q].min_max_tokens", modelID), StatusApplied, cur.MinMaxTokens, next.MinMaxTokens)
appendIfChanged(&changes, fmt.Sprintf("models[%q].default_thinking_token_budget", modelID), StatusApplied, cur.DefaultThinkingTokenBudget, next.DefaultThinkingTokenBudget)
appendDeepIfChanged(&changes, fmt.Sprintf("models[%q].providers", modelID), StatusApplied, cur.Providers, next.Providers)
}
for modelID := range candidateModels {
if _, exists := currentModels[modelID]; !exists {
changes = append(changes, Change{
Path: fmt.Sprintf("models[%q]", modelID),
Class: StatusApplied,
Previous: "absent",
Next: "present",
})
}
}
sort.SliceStable(changes, func(i, j int) bool {
if changes[i].Path == changes[j].Path {
return changes[i].Class < changes[j].Class
}
return changes[i].Path < changes[j].Path
})
// Determine overall status: any restart_required wins over applied.
hasRestart := false
for _, c := range changes {
if c.Class == StatusRestartRequired {
hasRestart = true
break
}
}
overallStatus := StatusApplied
summary := "no changes detected"
if hasRestart {
overallStatus = StatusRestartRequired
summary = "some changes require a process restart"
} else if len(changes) > 0 {
summary = "all changes can be applied without restart"
}
result := Result{
Status: overallStatus,
Changes: changes,
Summary: summary,
}
deriveReport(&result)
return result
}
// deriveReport populates the ops-report fields (changed node/provider/model ids
// and restart-required paths) from the classified change list. All slices are
// initialized non-nil so the JSON response stays stable for ops assertions.
func deriveReport(r *Result) {
nodes := newStringSet()
providers := newStringSet()
models := newStringSet()
var restartPaths []string
for _, c := range r.Changes {
if c.Class == StatusRestartRequired {
restartPaths = append(restartPaths, c.Path)
}
// Provider paths are nested under nodes[].providers[...]; check them
// before the broader nodes[ prefix so they are attributed correctly.
if id, ok := extractBracketID(c.Path, "providers["); ok {
providers.add(id)
continue
}
if id, ok := extractBracketID(c.Path, "nodes["); ok {
nodes.add(id)
continue
}
if id, ok := extractBracketID(c.Path, "models["); ok {
models.add(id)
}
}
r.ChangedNodes = nodes.sorted()
r.ChangedProviders = providers.sorted()
r.ChangedModels = models.sorted()
sort.Strings(restartPaths)
if restartPaths == nil {
restartPaths = []string{}
}
r.RestartRequiredPaths = restartPaths
}
// extractBracketID returns the quoted id immediately following prefix in path,
// e.g. extractBracketID(`nodes["node-1"].alias`, "nodes[") returns "node-1".
func extractBracketID(path, prefix string) (string, bool) {
i := strings.Index(path, prefix)
if i < 0 {
return "", false
}
rest := path[i+len(prefix):]
if len(rest) == 0 || rest[0] != '"' {
return "", false
}
rest = rest[1:]
j := strings.IndexByte(rest, '"')
if j <= 0 {
return "", false
}
return rest[:j], true
}
func cloneStringMap(m map[string]string) map[string]string {
if m == nil {
return nil
}
out := make(map[string]string, len(m))
for k, v := range m {
out[k] = v
}
return out
}
type stringSet struct {
m map[string]struct{}
}
func newStringSet() stringSet { return stringSet{m: make(map[string]struct{})} }
func (s stringSet) add(v string) { s.m[v] = struct{}{} }
func (s stringSet) sorted() []string {
out := make([]string, 0, len(s.m))
for v := range s.m {
out = append(out, v)
}
sort.Strings(out)
return out
}
// Evaluate loads the candidate config from req.ConfigPath, validates it, and
// classifies changes against current. On success it also returns the loaded
// candidate so the caller can apply mutable changes in apply mode.
func Evaluate(_ context.Context, current *config.EdgeConfig, req Request) (Result, *config.EdgeConfig, error) {
if req.Mode != ModeDryRun && req.Mode != ModeApply {
return RejectedResult(req, fmt.Sprintf("invalid mode %q: must be dry_run or apply", req.Mode)), nil, nil
}
candidate, err := LoadCandidate(req.ConfigPath)
if err != nil {
return RejectedResult(req, fmt.Sprintf("load candidate config: %v", err)), nil, nil
}
result := Classify(current, candidate)
result.Mode = req.Mode
result.RequestID = req.RequestID
if result.Changes == nil {
result.Changes = []Change{}
}
return result, candidate, nil
}