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"
	"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 = "postgres://nomadcode:nomadcode@code-server-postgres:5432/iop-control-plane-local?sslmode=disable"
	if cfg.Database.URL != wantDatabase {
		t.Fatalf("database url: got %q want %q", cfg.Database.URL, wantDatabase)
	}
	const wantRedis = "redis://code-server-redis:6379/1"
	if cfg.Redis.URL != wantRedis {
		t.Fatalf("redis url: got %q want %q", cfg.Redis.URL, wantRedis)
	}
	const wantRedisKeyPrefix = "iop:control-plane:local:"
	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 != "redis://code-server-redis:6379/1" {
		t.Fatalf("redis url: got %q", cfg.Redis.URL)
	}
	if cfg.Redis.KeyPrefix != "iop:control-plane:local:" {
		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)
	}
}

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,
							},
						},
					},
				},
			}, 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)
		}

		// Ensure raw settings/secret_key is NOT leaked anywhere in the JSON body
		bodyStr := resp.Body.String()
		if strings.Contains(bodyStr, "super-secret-token") || strings.Contains(bodyStr, "secret_key") || strings.Contains(bodyStr, "secret_settings") {
			t.Fatalf("leaked secret settings in response: %s", bodyStr)
		}
	})

	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 dispatched []string
	sendCommand := func(edgeID, operation, targetSelector string, parameters map[string]string, timeout time.Duration) (*iop.EdgeCommandResponse, error) {
		dispatched = append(dispatched, edgeID)
		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)
		}
	}
}