iop/apps/control-plane/cmd/control-plane/main_test.go

package main

import (
	"encoding/json"
	"fmt"
	"net/http"
	"net/http/httptest"
	"os"
	"path/filepath"
	"strings"
	"sync"
	"sync/atomic"
	"testing"
	"time"

	proto_socket "git.toki-labs.com/toki/proto-socket/go"
	"google.golang.org/protobuf/types/known/structpb"

	"iop/apps/control-plane/internal/wire"
	iop "iop/proto/gen/iop"
)

func TestLoadConfigDatabaseDefaultsToLocal(t *testing.T) {
	t.Setenv("IOP_DATABASE_URL", "")
	t.Setenv("IOP_REDIS_URL", "")
	t.Setenv("IOP_REDIS_KEY_PREFIX", "")
	cfg, err := loadConfig("")
	if err != nil {
		t.Fatalf("load config: %v", err)
	}
	const wantDatabase = ""
	if cfg.Database.URL != wantDatabase {
		t.Fatalf("database url: got %q want %q", cfg.Database.URL, wantDatabase)
	}
	const wantRedis = ""
	if cfg.Redis.URL != wantRedis {
		t.Fatalf("redis url: got %q want %q", cfg.Redis.URL, wantRedis)
	}
	const wantRedisKeyPrefix = "iop:control-plane:"
	if cfg.Redis.KeyPrefix != wantRedisKeyPrefix {
		t.Fatalf("redis key prefix: got %q want %q", cfg.Redis.KeyPrefix, wantRedisKeyPrefix)
	}
}

func TestLoadConfigDatabaseFromYAML(t *testing.T) {
	t.Setenv("IOP_DATABASE_URL", "")
	t.Setenv("IOP_REDIS_URL", "")
	t.Setenv("IOP_REDIS_KEY_PREFIX", "")
	path := writeConfig(t, `
server:
  listen: "127.0.0.1:9080"
database:
  url: "postgres://user:pass@db:5432/iop-control-plane-local?sslmode=disable"
redis:
  url: "redis://cache:6379/1"
  key_prefix: "iop:test:"
`)

	cfg, err := loadConfig(path)
	if err != nil {
		t.Fatalf("load config: %v", err)
	}
	const want = "postgres://user:pass@db:5432/iop-control-plane-local?sslmode=disable"
	if cfg.Database.URL != want {
		t.Fatalf("database url: got %q want %q", cfg.Database.URL, want)
	}
	if cfg.Redis.URL != "redis://cache:6379/1" {
		t.Fatalf("redis url: got %q", cfg.Redis.URL)
	}
	if cfg.Redis.KeyPrefix != "iop:test:" {
		t.Fatalf("redis key prefix: got %q", cfg.Redis.KeyPrefix)
	}
}

func TestLoadConfigRepositoryLocalConfig(t *testing.T) {
	t.Setenv("IOP_DATABASE_URL", "")
	t.Setenv("IOP_REDIS_URL", "")
	t.Setenv("IOP_REDIS_KEY_PREFIX", "")
	t.Setenv("IOP_WIRE_LISTEN", "")
	t.Setenv("IOP_EDGE_WIRE_LISTEN", "")
	cfg, err := loadConfig("../../../../configs/control-plane.yaml")
	if err != nil {
		t.Fatalf("load config: %v", err)
	}
	if cfg.Redis.URL != "" {
		t.Fatalf("redis url: got %q", cfg.Redis.URL)
	}
	if cfg.Redis.KeyPrefix != "iop:control-plane:" {
		t.Fatalf("redis key prefix: got %q", cfg.Redis.KeyPrefix)
	}
	if cfg.Server.WireListen != "0.0.0.0:19080" {
		t.Fatalf("wire_listen: got %q", cfg.Server.WireListen)
	}
	if cfg.Server.EdgeWireListen != "0.0.0.0:19081" {
		t.Fatalf("edge_wire_listen: got %q", cfg.Server.EdgeWireListen)
	}
	if cfg.Server.Listen != "0.0.0.0:18000" {
		t.Fatalf("listen: got %q want %q", cfg.Server.Listen, "0.0.0.0:18000")
	}
}

func TestLoadConfigEnvOverrides(t *testing.T) {
	path := writeConfig(t, `
database:
  url: "postgres://user:pass@db:5432/iop-control-plane-local?sslmode=disable"
redis:
  url: "redis://cache:6379/1"
  key_prefix: "iop:control-plane:local:"
`)
	const wantDatabase = "postgres://user:pass@db:5432/iop-control-plane-dev?sslmode=disable"
	const wantRedis = "redis://redis:6379/2"
	const wantRedisKeyPrefix = "iop:control-plane:dev:"
	t.Setenv("IOP_DATABASE_URL", wantDatabase)
	t.Setenv("IOP_REDIS_URL", wantRedis)
	t.Setenv("IOP_REDIS_KEY_PREFIX", wantRedisKeyPrefix)

	cfg, err := loadConfig(path)
	if err != nil {
		t.Fatalf("load config: %v", err)
	}
	if cfg.Database.URL != wantDatabase {
		t.Fatalf("database url: got %q want %q", cfg.Database.URL, wantDatabase)
	}
	if cfg.Redis.URL != wantRedis {
		t.Fatalf("redis url: got %q want %q", cfg.Redis.URL, wantRedis)
	}
	if cfg.Redis.KeyPrefix != wantRedisKeyPrefix {
		t.Fatalf("redis key prefix: got %q want %q", cfg.Redis.KeyPrefix, wantRedisKeyPrefix)
	}
}

func TestDatabaseLogFieldsDoNotExposeCredentials(t *testing.T) {
	fields := databaseLogFields("postgres://user:secret@db.example:5432/iop-control-plane-local?sslmode=disable")
	var host, database string
	for _, field := range fields {
		switch field.Key {
		case "host":
			host = field.String
		case "database":
			database = field.String
		}
	}
	if host != "db.example:5432" {
		t.Fatalf("host field: got %q", host)
	}
	if database != "iop-control-plane-local" {
		t.Fatalf("database field: got %q", database)
	}
	if strings.Contains(fmt.Sprint(fields), "secret") {
		t.Fatal("secret leaked into log fields")
	}
}

func TestRedisLogFields(t *testing.T) {
	fields := redisLogFields("redis://:secret@cache.example:6379/2", "iop:control-plane:dev:")
	var host, database, keyPrefix string
	for _, field := range fields {
		switch field.Key {
		case "host":
			host = field.String
		case "database":
			database = field.String
		case "key_prefix":
			keyPrefix = field.String
		}
	}
	if host != "cache.example:6379" {
		t.Fatalf("host field: got %q", host)
	}
	if database != "2" {
		t.Fatalf("database field: got %q", database)
	}
	if keyPrefix != "iop:control-plane:dev:" {
		t.Fatalf("key_prefix field: got %q", keyPrefix)
	}
	if strings.Contains(fmt.Sprint(fields), "secret") {
		t.Fatal("secret leaked into log fields")
	}
}

func writeConfig(t *testing.T, body string) string {
	t.Helper()
	path := filepath.Join(t.TempDir(), "control-plane.yaml")
	if err := os.WriteFile(path, []byte(body), 0o600); err != nil {
		t.Fatalf("write config: %v", err)
	}
	return path
}

func TestLoadConfigWireListenEnvOverrides(t *testing.T) {
	path := writeConfig(t, `
server:
  listen: "0.0.0.0:9080"
  wire_listen: "0.0.0.0:19080"
`)
	const wantListen = "127.0.0.1:9081"
	const wantWireListen = "127.0.0.1:19081"
	t.Setenv("IOP_LISTEN", wantListen)
	t.Setenv("IOP_WIRE_LISTEN", wantWireListen)

	cfg, err := loadConfig(path)
	if err != nil {
		t.Fatalf("load config: %v", err)
	}
	if cfg.Server.Listen != wantListen {
		t.Fatalf("listen: got %q want %q", cfg.Server.Listen, wantListen)
	}
	if cfg.Server.WireListen != wantWireListen {
		t.Fatalf("wire_listen: got %q want %q", cfg.Server.WireListen, wantWireListen)
	}
}

func TestLoadConfigEdgeWireListenDefault(t *testing.T) {
	t.Setenv("IOP_EDGE_WIRE_LISTEN", "")
	cfg, err := loadConfig("")
	if err != nil {
		t.Fatalf("load config: %v", err)
	}
	if cfg.Server.EdgeWireListen != "0.0.0.0:19081" {
		t.Fatalf("edge_wire_listen default: got %q want %q", cfg.Server.EdgeWireListen, "0.0.0.0:19081")
	}
	if cfg.Server.EdgeWireListen == cfg.Server.WireListen {
		t.Fatalf("edge_wire_listen must not share the client wire_listen address %q", cfg.Server.WireListen)
	}
}

func TestLoadConfigEdgeWireListenEnvOverrides(t *testing.T) {
	path := writeConfig(t, `
server:
  listen: "0.0.0.0:9080"
  wire_listen: "0.0.0.0:19080"
  edge_wire_listen: "0.0.0.0:19081"
`)
	const wantEdgeWireListen = "127.0.0.1:29081"
	t.Setenv("IOP_EDGE_WIRE_LISTEN", wantEdgeWireListen)

	cfg, err := loadConfig(path)
	if err != nil {
		t.Fatalf("load config: %v", err)
	}
	if cfg.Server.EdgeWireListen != wantEdgeWireListen {
		t.Fatalf("edge_wire_listen: got %q want %q", cfg.Server.EdgeWireListen, wantEdgeWireListen)
	}
	if cfg.Server.WireListen != "0.0.0.0:19080" {
		t.Fatalf("wire_listen should stay from YAML: got %q", cfg.Server.WireListen)
	}
}

func TestEdgeRegistryHTTPHandlersListAndGetEdge(t *testing.T) {
	registry := wire.NewEdgeRegistry()
	at := time.Unix(1780142200, 123).UTC()
	registry.MarkConnected(&iop.EdgeHelloRequest{
		EdgeId:       "edge-b",
		EdgeName:     "Edge B",
		Version:      "1.2.3",
		Capabilities: []string{"run", "node-registry"},
	}, at)
	registry.MarkConnected(&iop.EdgeHelloRequest{
		EdgeId:   "edge-a",
		EdgeName: "Edge A",
		Version:  "1.0.0",
	}, at.Add(time.Second))

	mux := http.NewServeMux()
	registerEdgeRegistryHandlers(mux, registry, nil, nil)

	listResp := httptest.NewRecorder()
	mux.ServeHTTP(listResp, httptest.NewRequest(http.MethodGet, "/edges", nil))
	if listResp.Code != http.StatusOK {
		t.Fatalf("GET /edges status=%d body=%s", listResp.Code, listResp.Body.String())
	}
	var list edgeRegistryResponse
	if err := json.Unmarshal(listResp.Body.Bytes(), &list); err != nil {
		t.Fatalf("decode list response: %v", err)
	}
	if len(list.Edges) != 2 {
		t.Fatalf("expected 2 edges, got %d", len(list.Edges))
	}
	if list.Edges[0].EdgeID != "edge-a" || list.Edges[1].EdgeID != "edge-b" {
		t.Fatalf("expected stable edge_id ordering, got %+v", list.Edges)
	}
	if list.Edges[1].Protocol != wire.Protocol {
		t.Fatalf("protocol: got %q want %q", list.Edges[1].Protocol, wire.Protocol)
	}
	if !list.Edges[1].Connected {
		t.Fatal("expected edge-b to be connected")
	}

	getResp := httptest.NewRecorder()
	mux.ServeHTTP(getResp, httptest.NewRequest(http.MethodGet, "/edges/edge-b", nil))
	if getResp.Code != http.StatusOK {
		t.Fatalf("GET /edges/edge-b status=%d body=%s", getResp.Code, getResp.Body.String())
	}
	var got edgeRegistryView
	if err := json.Unmarshal(getResp.Body.Bytes(), &got); err != nil {
		t.Fatalf("decode get response: %v", err)
	}
	if got.EdgeName != "Edge B" || got.Version != "1.2.3" {
		t.Fatalf("unexpected edge view: %+v", got)
	}
	if len(got.Capabilities) != 2 || got.Capabilities[0] != "run" || got.Capabilities[1] != "node-registry" {
		t.Fatalf("capabilities: got %+v", got.Capabilities)
	}
	if !got.LastSeen.Equal(at) {
		t.Fatalf("last_seen: got %s want %s", got.LastSeen, at)
	}
}

func TestEdgeRegistryHTTPHandlersListNodeEvents(t *testing.T) {
	registry := wire.NewEdgeRegistry()
	connectedAt := time.Unix(1780142200, 0).UTC()
	registry.MarkConnected(&iop.EdgeHelloRequest{
		EdgeId:   "edge-a",
		EdgeName: "Edge A",
		Version:  "1.0.0",
	}, connectedAt)
	registry.MarkConnected(&iop.EdgeHelloRequest{
		EdgeId:   "edge-b",
		EdgeName: "Edge B",
		Version:  "1.0.0",
	}, connectedAt)

	eventAt := time.Unix(1780142300, 0).UTC()
	receivedAt := eventAt.Add(time.Second)
	registry.RecordNodeEvent("edge-a", &iop.EdgeNodeEvent{
		EventId:   "evt-node-1",
		Type:      "node.connected",
		Source:    "edge",
		NodeId:    "node-1",
		Alias:     "alpha",
		Reason:    "registered",
		Timestamp: eventAt.UnixNano(),
		Metadata:  map[string]string{"rack": "r1"},
	}, receivedAt)
	registry.RecordNodeEvent("edge-b", &iop.EdgeNodeEvent{
		EventId:   "evt-node-2",
		Type:      "node.connected",
		NodeId:    "node-2",
		Timestamp: eventAt.UnixNano(),
	}, receivedAt)

	mux := http.NewServeMux()
	registerEdgeRegistryHandlers(mux, registry, nil, nil)

	resp := httptest.NewRecorder()
	mux.ServeHTTP(resp, httptest.NewRequest(http.MethodGet, "/edges/edge-a/events", nil))
	if resp.Code != http.StatusOK {
		t.Fatalf("GET /edges/edge-a/events status=%d body=%s", resp.Code, resp.Body.String())
	}
	var got edgeNodeEventsResponse
	if err := json.Unmarshal(resp.Body.Bytes(), &got); err != nil {
		t.Fatalf("decode events response: %v", err)
	}
	if got.EdgeID != "edge-a" {
		t.Fatalf("edge_id: got %q want edge-a", got.EdgeID)
	}
	if len(got.Events) != 1 {
		t.Fatalf("events len: got %d want 1", len(got.Events))
	}
	event := got.Events[0]
	if event.EventID != "evt-node-1" || event.Type != "node.connected" {
		t.Fatalf("unexpected event identity: %+v", event)
	}
	if event.NodeID != "node-1" || event.Alias != "alpha" || event.Reason != "registered" {
		t.Fatalf("unexpected node event view: %+v", event)
	}
	if !event.Timestamp.Equal(eventAt) {
		t.Fatalf("timestamp: got %s want %s", event.Timestamp, eventAt)
	}
	if !event.ReceivedAt.Equal(receivedAt) {
		t.Fatalf("received_at: got %s want %s", event.ReceivedAt, receivedAt)
	}
	if event.Metadata["rack"] != "r1" {
		t.Fatalf("metadata rack: got %q", event.Metadata["rack"])
	}

	filtered := httptest.NewRecorder()
	mux.ServeHTTP(filtered, httptest.NewRequest(http.MethodGet, "/edges/edge-a/events?node_id=missing", nil))
	if filtered.Code != http.StatusOK {
		t.Fatalf("filtered events status=%d body=%s", filtered.Code, filtered.Body.String())
	}
	var filteredGot edgeNodeEventsResponse
	if err := json.Unmarshal(filtered.Body.Bytes(), &filteredGot); err != nil {
		t.Fatalf("decode filtered events response: %v", err)
	}
	if len(filteredGot.Events) != 0 {
		t.Fatalf("expected empty filtered events, got %+v", filteredGot.Events)
	}

	extraResp := httptest.NewRecorder()
	mux.ServeHTTP(extraResp, httptest.NewRequest(http.MethodGet, "/edges/edge-a/events/extra", nil))
	if extraResp.Code != http.StatusNotFound {
		t.Fatalf("GET /edges/edge-a/events/extra status=%d body=%s", extraResp.Code, extraResp.Body.String())
	}

	missingResp := httptest.NewRecorder()
	mux.ServeHTTP(missingResp, httptest.NewRequest(http.MethodGet, "/edges/missing-edge/events", nil))
	if missingResp.Code != http.StatusNotFound {
		t.Fatalf("GET /edges/missing-edge/events status=%d body=%s", missingResp.Code, missingResp.Body.String())
	}
}

func TestEdgeRegistryHTTPHandlersRejectUnsupportedCases(t *testing.T) {
	registry := wire.NewEdgeRegistry()
	mux := http.NewServeMux()
	registerEdgeRegistryHandlers(mux, registry, nil, nil)

	postResp := httptest.NewRecorder()
	mux.ServeHTTP(postResp, httptest.NewRequest(http.MethodPost, "/edges", nil))
	if postResp.Code != http.StatusMethodNotAllowed {
		t.Fatalf("POST /edges status=%d body=%s", postResp.Code, postResp.Body.String())
	}

	missingResp := httptest.NewRecorder()
	mux.ServeHTTP(missingResp, httptest.NewRequest(http.MethodGet, "/edges/missing-edge", nil))
	if missingResp.Code != http.StatusNotFound {
		t.Fatalf("GET /edges/missing-edge status=%d body=%s", missingResp.Code, missingResp.Body.String())
	}
}

func TestEdgeRegistryHTTPHandlersGetLiveStatus(t *testing.T) {
	registry := wire.NewEdgeRegistry()
	connectedAt := time.Unix(1780142200, 0).UTC()
	registry.MarkConnected(&iop.EdgeHelloRequest{
		EdgeId:   "edge-a",
		EdgeName: "Edge A",
		Version:  "1.0.0",
	}, connectedAt)

	t.Run("Success", func(t *testing.T) {
		mux := http.NewServeMux()
		// Mock config payload containing secret settings to verify secrecy
		dummySettings, err := structpb.NewStruct(map[string]any{
			"secret_key": "super-secret-token",
		})
		if err != nil {
			t.Fatalf("create dummy settings: %v", err)
		}

		requestStatus := func(edgeID string, timeout time.Duration) (*iop.EdgeStatusResponse, error) {
			if edgeID != "edge-a" {
				return nil, fmt.Errorf("unexpected edge: %s", edgeID)
			}
			return &iop.EdgeStatusResponse{
				RequestId:            "req-123",
				EdgeId:               "edge-a",
				EdgeName:             "Edge A",
				ObservedTimeUnixNano: connectedAt.UnixNano(),
				Nodes: []*iop.EdgeNodeSnapshot{
					{
						NodeId:    "node-1",
						Alias:     "alpha",
						Label:     "node0",
						Connected: true,
						Config: &iop.NodeConfigPayload{
							Adapters: []*iop.AdapterConfig{
								{
									Type:     "ollama",
									Enabled:  true,
									Settings: dummySettings,
								},
							},
							Runtime: &iop.NodeRuntimeConfig{
								Concurrency: 4,
							},
						},
						ProviderSnapshots: []*iop.ProviderSnapshot{
							{
								Adapter:               "vllm-gpu",
								Status:                "available",
								Capacity:              4,
								InFlight:              1,
								Queued:                2,
								Id:                    "provider-vllm-primary",
								Type:                  "vllm",
								Category:              "api",
								ServedModels:          []string{"nvidia/Qwen3.6-35B-A3B-NVFP4"},
								Health:                "available",
								LoadRatio:             0.25,
								LifecycleCapabilities: []string{"list_models", "load_model"},
							},
							{
								Adapter:  "ollama-cpu",
								Status:   "backlog",
								Capacity: 2,
								InFlight: 2,
								Queued:   5,
								Id:       "provider-ollama-default",
								Type:     "ollama",
								Health:   "unavailable",
							},
						},
					},
				},
			}, nil
		}

		registerEdgeRegistryHandlers(mux, registry, requestStatus, nil)

		resp := httptest.NewRecorder()
		mux.ServeHTTP(resp, httptest.NewRequest(http.MethodGet, "/edges/edge-a/status", nil))

		if resp.Code != http.StatusOK {
			t.Fatalf("GET /edges/edge-a/status status=%d body=%s", resp.Code, resp.Body.String())
		}

		var view edgeStatusResponseView
		if err := json.Unmarshal(resp.Body.Bytes(), &view); err != nil {
			t.Fatalf("decode status view: %v", err)
		}

		if view.EdgeID != "edge-a" || len(view.Nodes) != 1 {
			t.Fatalf("unexpected view contents: %+v", view)
		}

		node := view.Nodes[0]
		if node.NodeID != "node-1" || node.Alias != "alpha" || !node.Connected {
			t.Fatalf("unexpected node state: %+v", node)
		}

		if node.Config == nil {
			t.Fatal("expected config summary to be populated")
		}

		if node.Config.Concurrency != 4 || len(node.Config.Adapters) != 1 {
			t.Fatalf("unexpected config summary: %+v", node.Config)
		}

		adapter := node.Config.Adapters[0]
		if adapter.Type != "ollama" || !adapter.Enabled {
			t.Fatalf("unexpected adapter summary: %+v", adapter)
		}

		// Provider snapshots assertions
		if len(node.ProviderSnapshots) != 2 {
			t.Fatalf("expected 2 provider snapshots, got %d", len(node.ProviderSnapshots))
		}

		// First provider snapshot (vllm-gpu)
		p0 := node.ProviderSnapshots[0]
		if p0.Adapter != "vllm-gpu" || p0.Status != "available" || p0.Capacity != 4 {
			t.Fatalf("unexpected provider snapshot[0]: %+v", p0)
		}
		if p0.InFlight != 1 || p0.Queued != 2 {
			t.Fatalf("unexpected in_flight/queued: in_flight=%d queued=%d", p0.InFlight, p0.Queued)
		}
		if p0.ID != "provider-vllm-primary" || p0.Type != "vllm" || p0.Category != "api" {
			t.Fatalf("unexpected id/type/category: id=%q type=%q category=%q", p0.ID, p0.Type, p0.Category)
		}
		if len(p0.ServedModels) != 1 || p0.ServedModels[0] != "nvidia/Qwen3.6-35B-A3B-NVFP4" {
			t.Fatalf("unexpected served_models: %+v", p0.ServedModels)
		}
		if p0.Health != "available" || p0.LoadRatio != 0.25 {
			t.Fatalf("unexpected health/load_ratio: health=%q load_ratio=%f", p0.Health, p0.LoadRatio)
		}
		if len(p0.LifecycleCapabilities) != 2 || p0.LifecycleCapabilities[0] != "list_models" || p0.LifecycleCapabilities[1] != "load_model" {
			t.Fatalf("unexpected lifecycle_capabilities: %+v", p0.LifecycleCapabilities)
		}

		// Second provider snapshot (ollama-cpu)
		p1 := node.ProviderSnapshots[1]
		if p1.Adapter != "ollama-cpu" || p1.Status != "backlog" || p1.Health != "unavailable" {
			t.Fatalf("unexpected provider snapshot[1]: %+v", p1)
		}
		if p1.Capacity != 2 || p1.InFlight != 2 || p1.Queued != 5 {
			t.Fatalf("unexpected capacity/in_flight/queued for ollama: capacity=%d in_flight=%d queued=%d", p1.Capacity, p1.InFlight, p1.Queued)
		}

		bodyStr := resp.Body.String()
		if strings.Contains(bodyStr, "super-secret-token") || strings.Contains(bodyStr, "secret_key") || strings.Contains(bodyStr, "settings") {
			t.Fatalf("leaked raw config settings in response: %s", bodyStr)
		}

		var raw map[string]any
		if err := json.Unmarshal(resp.Body.Bytes(), &raw); err != nil {
			t.Fatalf("decode raw status response: %v", err)
		}
		nodes, ok := raw["nodes"].([]any)
		if !ok || len(nodes) != 1 {
			t.Fatalf("raw nodes: got %#v", raw["nodes"])
		}
		rawNode, ok := nodes[0].(map[string]any)
		if !ok {
			t.Fatalf("raw node: got %#v", nodes[0])
		}
		rawConfig, ok := rawNode["config"].(map[string]any)
		if !ok {
			t.Fatalf("raw config: got %#v", rawNode["config"])
		}
		for _, forbidden := range []string{"runtime", "settings", "cli", "ollama", "vllm", "base_url", "args", "env", "resume_args"} {
			if _, exists := rawConfig[forbidden]; exists {
				t.Fatalf("config summary exposed raw key %q: %#v", forbidden, rawConfig)
			}
		}
		for _, required := range []string{"adapters", "concurrency"} {
			if _, exists := rawConfig[required]; !exists {
				t.Fatalf("config summary missing key %q: %#v", required, rawConfig)
			}
		}

		// Provider snapshots raw JSON assertions
		rawProviderSnapshots, ok := rawNode["provider_snapshots"].([]any)
		if !ok {
			t.Fatalf("provider_snapshots key missing or not an array: %#v", rawNode)
		}
		if len(rawProviderSnapshots) != 2 {
			t.Fatalf("expected 2 provider_snapshots in raw JSON, got %d", len(rawProviderSnapshots))
		}

		rawP0, ok := rawProviderSnapshots[0].(map[string]any)
		if !ok {
			t.Fatalf("provider_snapshots[0] is not an object: %#v", rawProviderSnapshots[0])
		}
		if rawP0["adapter"] != "vllm-gpu" {
			t.Fatalf("provider_snapshots[0].adapter: got %v", rawP0["adapter"])
		}
		if rawP0["capacity"] != float64(4) {
			t.Fatalf("provider_snapshots[0].capacity: got %v", rawP0["capacity"])
		}
		if rawP0["load_ratio"] != float64(0.25) {
			t.Fatalf("provider_snapshots[0].load_ratio: got %v", rawP0["load_ratio"])
		}
		servedModels, ok := rawP0["served_models"].([]any)
		if !ok || len(servedModels) != 1 {
			t.Fatalf("provider_snapshots[0].served_models: got %#v", rawP0["served_models"])
		}
		if servedModels[0] != "nvidia/Qwen3.6-35B-A3B-NVFP4" {
			t.Fatalf("provider_snapshots[0].served_models[0]: got %v", servedModels[0])
		}
		lifecycleCaps, ok := rawP0["lifecycle_capabilities"].([]any)
		if !ok || len(lifecycleCaps) != 2 {
			t.Fatalf("provider_snapshots[0].lifecycle_capabilities: got %#v", rawP0["lifecycle_capabilities"])
		}
		if lifecycleCaps[0] != "list_models" || lifecycleCaps[1] != "load_model" {
			t.Fatalf("provider_snapshots[0].lifecycle_capabilities: got %v", lifecycleCaps)
		}

		// Second provider snapshot raw check
		rawP1, ok := rawProviderSnapshots[1].(map[string]any)
		if !ok {
			t.Fatalf("provider_snapshots[1] is not an object: %#v", rawProviderSnapshots[1])
		}
		if rawP1["adapter"] != "ollama-cpu" || rawP1["health"] != "unavailable" {
			t.Fatalf("provider_snapshots[1] adapter/health: got %#v", rawP1)
		}
		// Zero load_ratio must be present in raw JSON (omitempty 제거 검증)
		if rawP1["load_ratio"] != float64(0) {
			t.Fatalf("provider_snapshots[1].load_ratio: expected 0 for idle provider, got %v", rawP1["load_ratio"])
		}

		// Ensure raw JSON does not leak config internals into provider_snapshots
		providerSnapshotsStr, _ := json.Marshal(rawP0)
		for _, forbidden := range []string{"runtime", "settings", "secret"} {
			if strings.Contains(string(providerSnapshotsStr), forbidden) {
				t.Fatalf("provider_snapshot leaked raw config key %q: %s", forbidden, providerSnapshotsStr)
			}
		}
	})

	t.Run("MissingEdge", func(t *testing.T) {
		mux := http.NewServeMux()
		requestStatus := func(edgeID string, timeout time.Duration) (*iop.EdgeStatusResponse, error) {
			return nil, fmt.Errorf("should not be called")
		}
		registerEdgeRegistryHandlers(mux, registry, requestStatus, nil)

		resp := httptest.NewRecorder()
		mux.ServeHTTP(resp, httptest.NewRequest(http.MethodGet, "/edges/missing-edge/status", nil))

		if resp.Code != http.StatusNotFound {
			t.Fatalf("expected 404 for missing edge, got status=%d", resp.Code)
		}
	})

	t.Run("UpstreamErrorOrTimeout", func(t *testing.T) {
		mux := http.NewServeMux()
		requestStatus := func(edgeID string, timeout time.Duration) (*iop.EdgeStatusResponse, error) {
			return nil, fmt.Errorf("upstream timeout")
		}
		registerEdgeRegistryHandlers(mux, registry, requestStatus, nil)

		resp := httptest.NewRecorder()
		mux.ServeHTTP(resp, httptest.NewRequest(http.MethodGet, "/edges/edge-a/status", nil))

		if resp.Code != http.StatusBadGateway {
			t.Fatalf("expected 502 for upstream error, got status=%d body=%s", resp.Code, resp.Body.String())
		}

		var errorResp map[string]string
		if err := json.Unmarshal(resp.Body.Bytes(), &errorResp); err != nil {
			t.Fatalf("decode error response: %v", err)
		}
		if errorResp["error"] != "upstream timeout" {
			t.Fatalf("unexpected error response: %+v", errorResp)
		}
	})
}

func TestEdgeCommandHTTPHandlerDispatchesToTargetEdge(t *testing.T) {
	registry := wire.NewEdgeRegistry()
	registry.MarkConnected(&iop.EdgeHelloRequest{EdgeId: "edge-a", EdgeName: "Edge A"}, time.Unix(1780142200, 0).UTC())

	t.Run("Dispatch", func(t *testing.T) {
		var (
			gotEdge      string
			gotOperation string
			gotSelector  string
			gotParams    map[string]string
		)
		sendCommand := func(edgeID, operation, targetSelector string, parameters map[string]string, timeout time.Duration) (*iop.EdgeCommandResponse, error) {
			gotEdge = edgeID
			gotOperation = operation
			gotSelector = targetSelector
			gotParams = parameters
			return &iop.EdgeCommandResponse{
				RequestId: "req-1",
				CommandId: "cmd-1",
				EdgeId:    edgeID,
				Status:    "ok",
				Summary:   "applied",
			}, nil
		}
		mux := http.NewServeMux()
		registerEdgeRegistryHandlers(mux, registry, nil, sendCommand)

		body := strings.NewReader(`{"operation":"restart-node","target_selector":"node-1","parameters":{"force":"true"}}`)
		resp := httptest.NewRecorder()
		mux.ServeHTTP(resp, httptest.NewRequest(http.MethodPost, "/edges/edge-a/commands", body))

		if resp.Code != http.StatusAccepted {
			t.Fatalf("POST commands status=%d body=%s", resp.Code, resp.Body.String())
		}
		if gotEdge != "edge-a" || gotOperation != "restart-node" || gotSelector != "node-1" {
			t.Fatalf("unexpected dispatch args: edge=%q op=%q selector=%q", gotEdge, gotOperation, gotSelector)
		}
		if gotParams["force"] != "true" {
			t.Fatalf("unexpected parameters: %+v", gotParams)
		}
		var view edgeCommandResponseView
		if err := json.Unmarshal(resp.Body.Bytes(), &view); err != nil {
			t.Fatalf("decode command response: %v", err)
		}
		if view.CommandID != "cmd-1" || view.EdgeID != "edge-a" || view.Status != "ok" {
			t.Fatalf("unexpected command view: %+v", view)
		}
	})

	t.Run("MissingOperation", func(t *testing.T) {
		mux := http.NewServeMux()
		registerEdgeRegistryHandlers(mux, registry, nil, func(string, string, string, map[string]string, time.Duration) (*iop.EdgeCommandResponse, error) {
			t.Fatal("sendCommand should not be called without an operation")
			return nil, nil
		})
		resp := httptest.NewRecorder()
		mux.ServeHTTP(resp, httptest.NewRequest(http.MethodPost, "/edges/edge-a/commands", strings.NewReader(`{}`)))
		if resp.Code != http.StatusBadRequest {
			t.Fatalf("expected 400 for missing operation, got %d", resp.Code)
		}
	})

	t.Run("OfflineEdgeReturnsError", func(t *testing.T) {
		sendCommand := func(string, string, string, map[string]string, time.Duration) (*iop.EdgeCommandResponse, error) {
			return nil, fmt.Errorf("edge \"edge-a\" is not connected")
		}
		mux := http.NewServeMux()
		registerEdgeRegistryHandlers(mux, registry, nil, sendCommand)

		resp := httptest.NewRecorder()
		mux.ServeHTTP(resp, httptest.NewRequest(http.MethodPost, "/edges/edge-a/commands", strings.NewReader(`{"operation":"drain"}`)))
		if resp.Code != http.StatusBadGateway {
			t.Fatalf("expected 502 for offline edge, got %d body=%s", resp.Code, resp.Body.String())
		}
		var errResp map[string]string
		if err := json.Unmarshal(resp.Body.Bytes(), &errResp); err != nil {
			t.Fatalf("decode error: %v", err)
		}
		if !strings.Contains(errResp["error"], "not connected") {
			t.Fatalf("unexpected error: %+v", errResp)
		}
	})

	t.Run("GetMethodNotAllowed", func(t *testing.T) {
		mux := http.NewServeMux()
		registerEdgeRegistryHandlers(mux, registry, nil, nil)
		resp := httptest.NewRecorder()
		mux.ServeHTTP(resp, httptest.NewRequest(http.MethodGet, "/edges/edge-a/commands", nil))
		if resp.Code != http.StatusMethodNotAllowed {
			t.Fatalf("expected 405 for GET on commands, got %d", resp.Code)
		}
	})
}

func TestEdgeOperationsHTTPHandlerListsAudit(t *testing.T) {
	registry := wire.NewEdgeRegistry()
	at := time.Unix(1780142200, 0).UTC()
	registry.MarkConnected(&iop.EdgeHelloRequest{EdgeId: "edge-a", EdgeName: "Edge A"}, at)
	registry.MarkConnected(&iop.EdgeHelloRequest{EdgeId: "edge-b", EdgeName: "Edge B"}, at)

	// Edge A command records
	registry.RecordCommandRequest("edge-a", &iop.EdgeCommandRequest{CommandId: "cmd-a-1", Operation: "restart-node", TargetSelector: "node-1"}, at)
	registry.RecordCommandEvent("edge-a", &iop.EdgeCommandEvent{CommandId: "cmd-a-1", Phase: "running", Summary: "executing"}, at.Add(time.Second))
	registry.RecordCommandResult("edge-a", &iop.EdgeCommandResponse{CommandId: "cmd-a-1", Status: "ok", Summary: "done"}, at.Add(2*time.Second))

	// Edge B command records
	registry.RecordCommandRequest("edge-b", &iop.EdgeCommandRequest{CommandId: "cmd-b-1", Operation: "drain-node", TargetSelector: "node-2"}, at)
	registry.RecordCommandEvent("edge-b", &iop.EdgeCommandEvent{CommandId: "cmd-b-1", Phase: "draining", Summary: "evicting pods"}, at.Add(time.Second))
	registry.RecordCommandResult("edge-b", &iop.EdgeCommandResponse{CommandId: "cmd-b-1", Status: "failed", Summary: "timeout"}, at.Add(2*time.Second))

	mux := http.NewServeMux()
	registerEdgeRegistryHandlers(mux, registry, nil, nil)

	// Request Edge A operations
	respA := httptest.NewRecorder()
	mux.ServeHTTP(respA, httptest.NewRequest(http.MethodGet, "/edges/edge-a/operations", nil))
	if respA.Code != http.StatusOK {
		t.Fatalf("GET /edges/edge-a/operations status=%d body=%s", respA.Code, respA.Body.String())
	}
	var gotA edgeOperationsResponse
	if err := json.Unmarshal(respA.Body.Bytes(), &gotA); err != nil {
		t.Fatalf("decode operations A: %v", err)
	}
	if gotA.EdgeID != "edge-a" || len(gotA.Operations) != 1 {
		t.Fatalf("unexpected operations A response: %+v", gotA)
	}
	opA := gotA.Operations[0]
	if opA.CommandID != "cmd-a-1" || opA.Operation != "restart-node" || opA.Status != "ok" || opA.Summary != "done" {
		t.Fatalf("unexpected operation A record: %+v", opA)
	}
	if len(opA.Events) != 1 || opA.Events[0].Phase != "running" {
		t.Fatalf("unexpected operation A events: %+v", opA.Events)
	}

	// Request Edge B operations
	respB := httptest.NewRecorder()
	mux.ServeHTTP(respB, httptest.NewRequest(http.MethodGet, "/edges/edge-b/operations", nil))
	if respB.Code != http.StatusOK {
		t.Fatalf("GET /edges/edge-b/operations status=%d body=%s", respB.Code, respB.Body.String())
	}
	var gotB edgeOperationsResponse
	if err := json.Unmarshal(respB.Body.Bytes(), &gotB); err != nil {
		t.Fatalf("decode operations B: %v", err)
	}
	if gotB.EdgeID != "edge-b" || len(gotB.Operations) != 1 {
		t.Fatalf("unexpected operations B response: %+v", gotB)
	}
	opB := gotB.Operations[0]
	if opB.CommandID != "cmd-b-1" || opB.Operation != "drain-node" || opB.Status != "failed" || opB.Summary != "timeout" {
		t.Fatalf("unexpected operation B record: %+v", opB)
	}
	if len(opB.Events) != 1 || opB.Events[0].Phase != "draining" {
		t.Fatalf("unexpected operation B events: %+v", opB.Events)
	}
}

func TestFleetStatusHTTPHandlerCombinesConnectionAndCapabilities(t *testing.T) {
	registry := wire.NewEdgeRegistry()
	at := time.Unix(1780142200, 0).UTC()
	registry.MarkConnected(&iop.EdgeHelloRequest{EdgeId: "edge-online", EdgeName: "Online", Version: "1.0.0"}, at)
	registry.MarkConnected(&iop.EdgeHelloRequest{EdgeId: "edge-other", EdgeName: "Other", Version: "1.1.0"}, at)
	registry.MarkConnected(&iop.EdgeHelloRequest{EdgeId: "edge-degraded", EdgeName: "Degraded"}, at)
	offlineToken := registry.MarkConnected(&iop.EdgeHelloRequest{EdgeId: "edge-offline", EdgeName: "Offline"}, at)
	registry.MarkDisconnected("edge-offline", offlineToken, at.Add(time.Second), proto_socket.DisconnectInfo{Reason: "closed"})

	requestStatus := func(edgeID string, timeout time.Duration) (*iop.EdgeStatusResponse, error) {
		switch edgeID {
		case "edge-online":
			return &iop.EdgeStatusResponse{
				EdgeId: edgeID,
				Nodes:  []*iop.EdgeNodeSnapshot{{NodeId: "node-1", Alias: "alpha", Connected: true}},
				Capabilities: []*iop.EdgeCapabilitySummary{
					{Kind: "run-dispatch", Available: true, Status: "ready"},
				},
				DomainAgents: []*iop.EdgeDomainAgentSummary{
					{AgentKind: "builder", Available: true, LifecycleState: "ready", ActiveCommandId: "cmd-builder-1", Summary: "builder is ready"},
				},
			}, nil
		case "edge-other":
			return &iop.EdgeStatusResponse{
				EdgeId: edgeID,
				Nodes:  []*iop.EdgeNodeSnapshot{{NodeId: "node-2", Alias: "beta", Connected: true}},
				Capabilities: []*iop.EdgeCapabilitySummary{
					{Kind: "run-dispatch", Available: true, Status: "ready"},
				},
				DomainAgents: []*iop.EdgeDomainAgentSummary{
					{AgentKind: "deployer", Available: true, LifecycleState: "busy", ActiveCommandId: "cmd-dep-2", Summary: "deployer is busy"},
				},
			}, nil
		case "edge-degraded":
			return nil, fmt.Errorf("status timeout")
		default:
			return nil, fmt.Errorf("unexpected status request for %q", edgeID)
		}
	}

	mux := http.NewServeMux()
	registerFleetHandlers(mux, registry, requestStatus, nil)

	resp := httptest.NewRecorder()
	mux.ServeHTTP(resp, httptest.NewRequest(http.MethodGet, "/fleet/status", nil))
	if resp.Code != http.StatusOK {
		t.Fatalf("GET /fleet/status status=%d body=%s", resp.Code, resp.Body.String())
	}
	var got fleetStatusResponse
	if err := json.Unmarshal(resp.Body.Bytes(), &got); err != nil {
		t.Fatalf("decode fleet status: %v", err)
	}
	if len(got.Edges) != 4 {
		t.Fatalf("expected 4 fleet edges, got %d", len(got.Edges))
	}
	health := map[string]fleetEdgeView{}
	for _, e := range got.Edges {
		health[e.EdgeID] = e
	}

	// Verify edge-online
	eOnline := health["edge-online"]
	if eOnline.Health != "online" || eOnline.NodeCount != 1 || len(eOnline.Capabilities) != 1 {
		t.Fatalf("unexpected online edge view: %+v", eOnline)
	}
	if len(eOnline.DomainAgents) != 1 || eOnline.DomainAgents[0].AgentKind != "builder" || eOnline.DomainAgents[0].LifecycleState != "ready" || eOnline.DomainAgents[0].ActiveCommandID != "cmd-builder-1" || eOnline.DomainAgents[0].Summary != "builder is ready" {
		t.Fatalf("unexpected domain agents for edge-online: %+v", eOnline.DomainAgents)
	}

	// Verify edge-other
	eOther := health["edge-other"]
	if eOther.Health != "online" || eOther.NodeCount != 1 || len(eOther.Capabilities) != 1 {
		t.Fatalf("unexpected edge-other view: %+v", eOther)
	}
	if len(eOther.DomainAgents) != 1 || eOther.DomainAgents[0].AgentKind != "deployer" || eOther.DomainAgents[0].LifecycleState != "busy" || eOther.DomainAgents[0].ActiveCommandID != "cmd-dep-2" || eOther.DomainAgents[0].Summary != "deployer is busy" {
		t.Fatalf("unexpected domain agents for edge-other: %+v", eOther.DomainAgents)
	}

	// Verify edge-degraded
	if e := health["edge-degraded"]; e.Health != "degraded" || e.Error == "" {
		t.Fatalf("unexpected degraded edge view: %+v", e)
	}

	// Verify edge-offline
	if e := health["edge-offline"]; e.Health != "offline" || e.Connected {
		t.Fatalf("unexpected offline edge view: %+v", e)
	}

	// Fleet status must not leak Node-direct internals (addresses, tokens) or
	// raw node snapshots; it surfaces a connection + capability summary only.
	bodyStr := resp.Body.String()
	for _, forbidden := range []string{"node_id", "token", "address", "\"nodes\""} {
		if strings.Contains(bodyStr, forbidden) {
			t.Fatalf("fleet status leaked %q: %s", forbidden, bodyStr)
		}
	}
}

func TestFleetCommandsHTTPHandlerFansOutToConnectedEdgesOnly(t *testing.T) {
	registry := wire.NewEdgeRegistry()
	at := time.Unix(1780142200, 0).UTC()
	registry.MarkConnected(&iop.EdgeHelloRequest{EdgeId: "edge-a"}, at)
	registry.MarkConnected(&iop.EdgeHelloRequest{EdgeId: "edge-b"}, at)
	registry.MarkConnected(&iop.EdgeHelloRequest{EdgeId: "edge-off"}, at)
	registry.MarkDisconnected("edge-off", 3, at.Add(time.Second), proto_socket.DisconnectInfo{Reason: "closed"})

	var dispatchMu sync.Mutex
	var dispatched []string
	sendCommand := func(edgeID, operation, targetSelector string, parameters map[string]string, timeout time.Duration) (*iop.EdgeCommandResponse, error) {
		dispatchMu.Lock()
		dispatched = append(dispatched, edgeID)
		dispatchMu.Unlock()
		return &iop.EdgeCommandResponse{CommandId: "cmd-" + edgeID, EdgeId: edgeID, Status: "ok"}, nil
	}

	mux := http.NewServeMux()
	registerFleetHandlers(mux, registry, nil, sendCommand)

	resp := httptest.NewRecorder()
	mux.ServeHTTP(resp, httptest.NewRequest(http.MethodPost, "/fleet/commands", strings.NewReader(`{"operation":"drain"}`)))
	if resp.Code != http.StatusAccepted {
		t.Fatalf("POST /fleet/commands status=%d body=%s", resp.Code, resp.Body.String())
	}
	var got fleetCommandResponse
	if err := json.Unmarshal(resp.Body.Bytes(), &got); err != nil {
		t.Fatalf("decode fleet command response: %v", err)
	}
	if got.Operation != "drain" || len(got.Results) != 2 {
		t.Fatalf("expected fan-out to 2 connected edges, got %+v", got)
	}
	if len(dispatched) != 2 {
		t.Fatalf("expected dispatch to 2 connected edges, got %+v", dispatched)
	}
	for _, edge := range dispatched {
		if edge == "edge-off" {
			t.Fatalf("command was dispatched to a disconnected edge: %+v", dispatched)
		}
	}
}

func TestHTTPMuxHealthAndReadiness(t *testing.T) {
	mux := newHTTPMux()

	t.Run("Health", func(t *testing.T) {
		resp := httptest.NewRecorder()
		mux.ServeHTTP(resp, httptest.NewRequest(http.MethodGet, "/healthz", nil))
		if resp.Code != http.StatusOK {
			t.Fatalf("GET /healthz status=%d, want %d", resp.Code, http.StatusOK)
		}
		if resp.Body.String() != "ok\n" {
			t.Fatalf("GET /healthz body=%q, want %q", resp.Body.String(), "ok\n")
		}
	})

	t.Run("Readiness", func(t *testing.T) {
		resp := httptest.NewRecorder()
		mux.ServeHTTP(resp, httptest.NewRequest(http.MethodGet, "/readyz", nil))
		if resp.Code != http.StatusOK {
			t.Fatalf("GET /readyz status=%d, want %d", resp.Code, http.StatusOK)
		}
		if resp.Body.String() != "ready\n" {
			t.Fatalf("GET /readyz body=%q, want %q", resp.Body.String(), "ready\n")
		}
	})
}

func TestHTTPMuxRegistersEdgeHandlers(t *testing.T) {
	registry := wire.NewEdgeRegistry()
	mux := newHTTPMux()
	registerEdgeRegistryHandlers(mux, registry, nil, nil)
	registerFleetHandlers(mux, registry, nil, nil)

	t.Run("ListEdges", func(t *testing.T) {
		resp := httptest.NewRecorder()
		mux.ServeHTTP(resp, httptest.NewRequest(http.MethodGet, "/edges", nil))
		if resp.Code != http.StatusOK {
			t.Fatalf("GET /edges status=%d", resp.Code)
		}
	})

	t.Run("FleetStatus", func(t *testing.T) {
		resp := httptest.NewRecorder()
		mux.ServeHTTP(resp, httptest.NewRequest(http.MethodGet, "/fleet/status", nil))
		if resp.Code != http.StatusOK {
			t.Fatalf("GET /fleet/status status=%d", resp.Code)
		}
	})
}

func TestFleetStatusHTTPHandlerUsesConfiguredTimeout(t *testing.T) {
	registry := wire.NewEdgeRegistry()
	at := time.Unix(1780142200, 0).UTC()
	registry.MarkConnected(&iop.EdgeHelloRequest{EdgeId: "edge-a"}, at)

	var gotTimeout time.Duration
	requestStatus := func(edgeID string, timeout time.Duration) (*iop.EdgeStatusResponse, error) {
		gotTimeout = timeout
		return &iop.EdgeStatusResponse{EdgeId: edgeID}, nil
	}

	mux := http.NewServeMux()
	registerFleetHandlersWithOptions(mux, registry, requestStatus, nil, fleetOptions{StatusTimeout: 1234 * time.Millisecond})

	resp := httptest.NewRecorder()
	mux.ServeHTTP(resp, httptest.NewRequest(http.MethodGet, "/fleet/status", nil))
	if resp.Code != http.StatusOK {
		t.Fatalf("GET /fleet/status status=%d body=%s", resp.Code, resp.Body.String())
	}
	if gotTimeout != 1234*time.Millisecond {
		t.Fatalf("requester timeout=%s, want %s", gotTimeout, 1234*time.Millisecond)
	}
}

func TestFleetStatusHTTPHandlerBoundsConcurrentStatusRequests(t *testing.T) {
	registry := wire.NewEdgeRegistry()
	at := time.Unix(1780142200, 0).UTC()
	for _, id := range []string{"edge-1", "edge-2", "edge-3", "edge-4", "edge-5"} {
		registry.MarkConnected(&iop.EdgeHelloRequest{EdgeId: id}, at)
	}

	var inFlight, maxInFlight int32
	requestStatus := func(edgeID string, timeout time.Duration) (*iop.EdgeStatusResponse, error) {
		cur := atomic.AddInt32(&inFlight, 1)
		for {
			old := atomic.LoadInt32(&maxInFlight)
			if cur <= old || atomic.CompareAndSwapInt32(&maxInFlight, old, cur) {
				break
			}
		}
		time.Sleep(20 * time.Millisecond)
		atomic.AddInt32(&inFlight, -1)
		return &iop.EdgeStatusResponse{EdgeId: edgeID}, nil
	}

	mux := http.NewServeMux()
	registerFleetHandlersWithOptions(mux, registry, requestStatus, nil, fleetOptions{MaxConcurrentStatus: 2})

	resp := httptest.NewRecorder()
	mux.ServeHTTP(resp, httptest.NewRequest(http.MethodGet, "/fleet/status", nil))
	if resp.Code != http.StatusOK {
		t.Fatalf("GET /fleet/status status=%d body=%s", resp.Code, resp.Body.String())
	}
	if got := atomic.LoadInt32(&maxInFlight); got > 2 {
		t.Fatalf("max in-flight status requests=%d, want <= 2", got)
	}

	var got fleetStatusResponse
	if err := json.Unmarshal(resp.Body.Bytes(), &got); err != nil {
		t.Fatalf("decode fleet status: %v", err)
	}
	wantOrder := []string{"edge-1", "edge-2", "edge-3", "edge-4", "edge-5"}
	if len(got.Edges) != len(wantOrder) {
		t.Fatalf("expected %d edges, got %d", len(wantOrder), len(got.Edges))
	}
	for i, want := range wantOrder {
		if got.Edges[i].EdgeID != want {
			t.Fatalf("edge[%d]=%q, want %q (response must keep snapshot order)", i, got.Edges[i].EdgeID, want)
		}
	}
}

func TestFleetStatusHTTPHandlerReusesFreshStatusCache(t *testing.T) {
	registry := wire.NewEdgeRegistry()
	at := time.Unix(1780142200, 0).UTC()
	registry.MarkConnected(&iop.EdgeHelloRequest{EdgeId: "edge-a"}, at)

	var calls int32
	requestStatus := func(edgeID string, timeout time.Duration) (*iop.EdgeStatusResponse, error) {
		atomic.AddInt32(&calls, 1)
		return &iop.EdgeStatusResponse{EdgeId: edgeID}, nil
	}

	mux := http.NewServeMux()
	registerFleetHandlersWithOptions(mux, registry, requestStatus, nil, fleetOptions{StatusCacheTTL: time.Minute})

	for i := 0; i < 2; i++ {
		resp := httptest.NewRecorder()
		mux.ServeHTTP(resp, httptest.NewRequest(http.MethodGet, "/fleet/status", nil))
		if resp.Code != http.StatusOK {
			t.Fatalf("GET /fleet/status #%d status=%d body=%s", i, resp.Code, resp.Body.String())
		}
	}
	if got := atomic.LoadInt32(&calls); got != 1 {
		t.Fatalf("requester call count=%d, want 1 (fresh cache must be reused)", got)
	}
}

func TestFleetCommandsHTTPHandlerBoundsConcurrentFanout(t *testing.T) {
	registry := wire.NewEdgeRegistry()
	at := time.Unix(1780142200, 0).UTC()
	for _, id := range []string{"edge-1", "edge-2", "edge-3", "edge-4", "edge-5"} {
		registry.MarkConnected(&iop.EdgeHelloRequest{EdgeId: id}, at)
	}
	registry.MarkConnected(&iop.EdgeHelloRequest{EdgeId: "edge-off"}, at)
	registry.MarkDisconnected("edge-off", 6, at.Add(time.Second), proto_socket.DisconnectInfo{Reason: "closed"})

	var inFlight, maxInFlight int32
	sendCommand := func(edgeID, operation, targetSelector string, parameters map[string]string, timeout time.Duration) (*iop.EdgeCommandResponse, error) {
		cur := atomic.AddInt32(&inFlight, 1)
		for {
			old := atomic.LoadInt32(&maxInFlight)
			if cur <= old || atomic.CompareAndSwapInt32(&maxInFlight, old, cur) {
				break
			}
		}
		time.Sleep(20 * time.Millisecond)
		atomic.AddInt32(&inFlight, -1)
		return &iop.EdgeCommandResponse{CommandId: "cmd-" + edgeID, EdgeId: edgeID, Status: "ok"}, nil
	}

	mux := http.NewServeMux()
	registerFleetHandlersWithOptions(mux, registry, nil, sendCommand, fleetOptions{MaxConcurrentCommands: 2})

	resp := httptest.NewRecorder()
	mux.ServeHTTP(resp, httptest.NewRequest(http.MethodPost, "/fleet/commands", strings.NewReader(`{"operation":"drain"}`)))
	if resp.Code != http.StatusAccepted {
		t.Fatalf("POST /fleet/commands status=%d body=%s", resp.Code, resp.Body.String())
	}
	if got := atomic.LoadInt32(&maxInFlight); got > 2 {
		t.Fatalf("max in-flight command dispatch=%d, want <= 2", got)
	}

	var got fleetCommandResponse
	if err := json.Unmarshal(resp.Body.Bytes(), &got); err != nil {
		t.Fatalf("decode fleet command response: %v", err)
	}
	wantOrder := []string{"edge-1", "edge-2", "edge-3", "edge-4", "edge-5"}
	if len(got.Results) != len(wantOrder) {
		t.Fatalf("expected fan-out to %d connected edges, got %d: %+v", len(wantOrder), len(got.Results), got.Results)
	}
	for i, want := range wantOrder {
		if got.Results[i].EdgeID != want {
			t.Fatalf("result[%d]=%q, want %q (results must keep connected snapshot order)", i, got.Results[i].EdgeID, want)
		}
		if got.Results[i].EdgeID == "edge-off" {
			t.Fatalf("command was dispatched to a disconnected edge: %+v", got.Results)
		}
	}
}

func TestFleetCommandsHTTPHandlerUsesConfiguredTimeout(t *testing.T) {
	registry := wire.NewEdgeRegistry()
	at := time.Unix(1780142200, 0).UTC()
	registry.MarkConnected(&iop.EdgeHelloRequest{EdgeId: "edge-a"}, at)

	var gotTimeout time.Duration
	sendCommand := func(edgeID, operation, targetSelector string, parameters map[string]string, timeout time.Duration) (*iop.EdgeCommandResponse, error) {
		gotTimeout = timeout
		return &iop.EdgeCommandResponse{CommandId: "cmd-" + edgeID, EdgeId: edgeID, Status: "ok"}, nil
	}

	mux := http.NewServeMux()
	registerFleetHandlersWithOptions(mux, registry, nil, sendCommand, fleetOptions{CommandTimeout: 4321 * time.Millisecond})

	resp := httptest.NewRecorder()
	mux.ServeHTTP(resp, httptest.NewRequest(http.MethodPost, "/fleet/commands", strings.NewReader(`{"operation":"drain"}`)))
	if resp.Code != http.StatusAccepted {
		t.Fatalf("POST /fleet/commands status=%d body=%s", resp.Code, resp.Body.String())
	}
	if gotTimeout != 4321*time.Millisecond {
		t.Fatalf("sendCommand timeout=%s, want %s", gotTimeout, 4321*time.Millisecond)
	}
}