feat(control-plane): edge registry and server refactoring

- Refactor edge_registry.go and edge_server.go - Update edge_server_test.go with new test cases - Update main.go and main_test.go for edge integration - Move completed tasks to archive
2026-06-04 07:42:57 +09:00 · 2026-06-04 07:42:57 +09:00 · f9917dd2fd
commit f9917dd2fd
parent 63cddb8a1e
9 changed files with 1125 additions and 88 deletions
--- a/agent-task/archive/2026/06/m-multi-edge-operations/03+01,02_control_plane_fleet_api/code_review_cloud_G06_0.log
+++ b/agent-task/archive/2026/06/m-multi-edge-operations/03+01,02_control_plane_fleet_api/code_review_cloud_G06_0.log
@ -0,0 +1,61 @@
+<!-- task=m-multi-edge-operations/03+01,02_control_plane_fleet_api review=0 tag=API -->
+# Code Review - Control Plane Fleet API
+
+## Review Scope
+
+- `apps/control-plane/internal/wire/edge_server.go`
+- `apps/control-plane/internal/wire/edge_registry.go`
+- `apps/control-plane/cmd/control-plane/main.go`
+- Control Plane API/wire tests
+
+## Dependency Check
+
+- [x] `agent-task/m-multi-edge-operations/01_native_contract/complete.log` 확인
+- [x] `agent-task/m-multi-edge-operations/02+01_edge_fleet_ops/complete.log` 확인
+
+## Reviewer Checklist
+
+- [x] Control Plane 저장 모델이 connection/control/audit view에 머문다.
+- [x] Edge 설정/Node registry/runtime 원본을 복제하지 않는다.
+- [x] command routing이 edge별 active client로 정확히 분리된다.
+- [x] offline/reconnect/timeout 응답이 명확하다.
+- [x] bounded audit/event 저장 정책이 있다.
+- [x] 기존 Edge 목록/status/events API가 회귀하지 않았다.
+
+## Verification Result
+
+- [x] `go test -count=1 ./apps/control-plane/internal/wire ./apps/control-plane/cmd/control-plane`
+- [x] `go test -count=1 ./apps/control-plane/...`
+- [x] `go test -count=1 ./apps/edge/internal/controlplane`
+- [x] `git diff --check`
+- [x] `make test-control-plane-edge-wire`
+
+## User Review
+
+검토 기본값: 이 리뷰는 사용자가 명시적으로 보류하지 않으면 구현자가 자체 검증 후 완료 처리한다.
+
+## 코드리뷰 결과
+
+- 종합 판정: PASS
+- 차원별 평가:
+  - Correctness: Pass
+  - Completeness: Pass
+  - Test coverage: Pass
+  - API contract: Pass
+  - Code quality: Pass
+  - Plan deviation: Pass
+  - Verification trust: Pass
+- 발견된 문제: 없음
+- 다음 단계: PASS이므로 active plan/review를 로그로 아카이브하고 `complete.log` 작성 후 task directory를 archive로 이동한다.
+
+## 코드리뷰 전용 체크리스트
+
+- [x] `코드리뷰 결과`에 `PASS`, `WARN`, `FAIL` 중 하나의 판정을 append했다.
+- [x] 판정과 `차원별 평가`, Required/Suggested/Nit 분류가 서로 일치한다.
+- [x] active `CODE_REVIEW-cloud-G06.md`를 `code_review_cloud_G06_0.log`로 아카이브했다.
+- [x] active `PLAN-cloud-G06.md`를 `plan_cloud_G06_0.log`로 아카이브했다.
+- [x] `.gitignore`의 Agent-Ops 관리 block이 `agent-task/**/*.md`와 `agent-task/**/*.log`를 unignore하고 `agent-roadmap/current.md`를 ignore하는지 확인했다.
+- [x] PASS이므로 `agent-ops/skills/common/code-review/templates/complete-log-template.md` 기준으로 `complete.log`를 작성하고 active `.md` 파일을 남기지 않았다.
+- [x] PASS이므로 active task 디렉터리 `agent-task/m-multi-edge-operations/03+01,02_control_plane_fleet_api/`를 `agent-task/archive/2026/06/m-multi-edge-operations/03+01,02_control_plane_fleet_api/`로 이동했다.
+- [x] PASS이고 task group이 `m-multi-edge-operations`이므로 런타임이 읽을 완료 이벤트 메타데이터를 보고하고, roadmap 수정이나 `update-roadmap` 직접 호출을 하지 않는다.
+- [x] PASS split 작업이며 active parent `agent-task/m-multi-edge-operations/`에는 sibling 작업이 남아 유지했다.
--- a/agent-task/archive/2026/06/m-multi-edge-operations/03+01,02_control_plane_fleet_api/complete.log
+++ b/agent-task/archive/2026/06/m-multi-edge-operations/03+01,02_control_plane_fleet_api/complete.log
@ -0,0 +1,39 @@
+# Complete - m-multi-edge-operations/03+01,02_control_plane_fleet_api
+
+## 완료 일시
+
+2026-06-04
+
+## 요약
+
+Control Plane Fleet API 구현을 1회 리뷰 루프로 검토했고 최종 판정은 PASS다.
+
+## 루프 이력
+
+| Plan | Review | Verdict | 메모 |
+|------|--------|---------|------|
+| `plan_cloud_G06_0.log` | `code_review_cloud_G06_0.log` | PASS | Multi-edge command API, command result/event relay, fleet status summary API, bounded audit 저장, edge별 command routing이 계획 범위를 충족했다. |
+
+## 구현/정리 내용
+
+- Control Plane HTTP API에 `/edges/{id}/status`, `/edges/{id}/operations`, `POST /edges/{id}/commands`, `/fleet/status`, `POST /fleet/commands`를 추가했다.
+- `EdgeServer.SendCommand`와 `EdgeRegistry` command audit/event view를 추가해 target Edge의 active client로만 command를 보내고 result/event를 bounded in-memory view로 보관한다.
+- API 응답은 connection/control/audit summary와 Edge-provided capability/domain-agent summary로 제한하고 Edge 설정, Node registry, runtime 원본 복제를 피했다.
+- command routing, offline error, command lifecycle event relay, bounded audit copy, fleet status/command fan-out, 기존 Edge list/status/events 회귀 테스트를 보강했다.
+
+## 최종 검증
+
+- `git diff --check` - PASS; 출력 없음.
+- `go test -count=1 ./apps/control-plane/internal/wire ./apps/control-plane/cmd/control-plane` - PASS; `iop/apps/control-plane/internal/wire` 1.494s, `iop/apps/control-plane/cmd/control-plane` 0.016s.
+- `go test -count=1 ./apps/control-plane/...` - PASS; `cmd/control-plane`와 `internal/wire` 통과.
+- `go test -count=1 ./apps/edge/internal/controlplane` - PASS; `iop/apps/edge/internal/controlplane` 4.453s.
+- `go test -count=1 ./apps/control-plane/internal/wire -run 'TestEdgeServer(SendsCommandToTargetEdgeOnly|RecordsCommandLifecycleEventsFromEdge|SendCommandFailsWhenEdgeNotConnected)|TestEdgeRegistryCommandsAreBoundedAndCopied'` - PASS; `iop/apps/control-plane/internal/wire` 0.187s.
+- `make test-control-plane-edge-wire` - PASS; Control Plane-Edge wire smoke에서 hello accepted, connected to control plane, edge disconnected marker를 확인했다.
+
+## 잔여 Nit
+
+- 없음
+
+## 후속 작업
+
+- 없음
--- a/agent-task/archive/2026/06/m-multi-edge-operations/03+01,02_control_plane_fleet_api/plan_cloud_G06_0.log
+++ b/agent-task/archive/2026/06/m-multi-edge-operations/03+01,02_control_plane_fleet_api/plan_cloud_G06_0.log
--- a/agent-task/m-multi-edge-operations/03+01,02_control_plane_fleet_api/CODE_REVIEW-cloud-G06.md
+++ b/agent-task/m-multi-edge-operations/03+01,02_control_plane_fleet_api/CODE_REVIEW-cloud-G06.md
@ -1,33 +0,0 @@
-<!-- task=m-multi-edge-operations/03+01,02_control_plane_fleet_api review=0 tag=API -->
-# Code Review - Control Plane Fleet API
-
-## Review Scope
-
- `apps/control-plane/internal/wire/edge_server.go`
- `apps/control-plane/internal/wire/edge_registry.go`
- `apps/control-plane/cmd/control-plane/main.go`
- Control Plane API/wire tests
-
-## Dependency Check
-
- [ ] `agent-task/m-multi-edge-operations/01_native_contract/complete.log` 확인
- [ ] `agent-task/m-multi-edge-operations/02+01_edge_fleet_ops/complete.log` 확인
-
-## Reviewer Checklist
-
- [ ] Control Plane 저장 모델이 connection/control/audit view에 머문다.
- [ ] Edge 설정/Node registry/runtime 원본을 복제하지 않는다.
- [ ] command routing이 edge별 active client로 정확히 분리된다.
- [ ] offline/reconnect/timeout 응답이 명확하다.
- [ ] bounded audit/event 저장 정책이 있다.
- [ ] 기존 Edge 목록/status/events API가 회귀하지 않았다.
-
-## Verification Result
-
- [ ] `go test ./apps/control-plane/internal/wire ./apps/control-plane/cmd/control-plane`
- [ ] 필요한 경우 `go test ./apps/control-plane/...`
- [ ] 존재 시 `make test-control-plane-edge-wire`
-
-## User Review
-
-검토 기본값: 이 리뷰는 사용자가 명시적으로 보류하지 않으면 구현자가 자체 검증 후 완료 처리한다.
--- a/apps/control-plane/cmd/control-plane/main.go
+++ b/apps/control-plane/cmd/control-plane/main.go
@ -179,7 +179,8 @@ func run(ctx context.Context, cfg controlPlaneConfig, logger *zap.Logger) error
 		return fmt.Errorf("start edge wire server: %w", err)
 	}
 	defer func() { _ = edgeServer.Stop() }()
-	registerEdgeRegistryHandlers(mux, edgeServer.Registry(), edgeServer.RequestStatus)
+	registerEdgeRegistryHandlers(mux, edgeServer.Registry(), edgeServer.RequestStatus, edgeServer.SendCommand)
+	registerFleetHandlers(mux, edgeServer.Registry(), edgeServer.RequestStatus, edgeServer.SendCommand)

 	server := &http.Server{
 		Addr:              cfg.Server.Listen,
@ -240,13 +241,96 @@ type edgeNodeEventView struct {
 }

 type edgeStatusResponseView struct {
-	RequestID            string                 `json:"request_id"`
-	EdgeID               string                 `json:"edge_id"`
-	EdgeName             string                 `json:"edge_name"`
-	ObservedTimeUnixNano int64                  `json:"observed_time_unix_nano"`
-	Nodes                []edgeNodeSnapshotView `json:"nodes"`
-	Metadata             map[string]string      `json:"metadata,omitempty"`
-	Error                string                 `json:"error,omitempty"`
+	RequestID            string                       `json:"request_id"`
+	EdgeID               string                       `json:"edge_id"`
+	EdgeName             string                       `json:"edge_name"`
+	ObservedTimeUnixNano int64                        `json:"observed_time_unix_nano"`
+	Nodes                []edgeNodeSnapshotView       `json:"nodes"`
+	Capabilities         []edgeCapabilitySummaryView  `json:"capabilities,omitempty"`
+	DomainAgents         []edgeDomainAgentSummaryView `json:"domain_agents,omitempty"`
+	Metadata             map[string]string            `json:"metadata,omitempty"`
+	Error                string                       `json:"error,omitempty"`
+}
+
+type edgeCapabilitySummaryView struct {
+	Kind      string `json:"kind"`
+	Available bool   `json:"available"`
+	Status    string `json:"status,omitempty"`
+	Summary   string `json:"summary,omitempty"`
+}
+
+type edgeDomainAgentSummaryView struct {
+	AgentKind       string `json:"agent_kind"`
+	Available       bool   `json:"available"`
+	LifecycleState  string `json:"lifecycle_state,omitempty"`
+	ActiveCommandID string `json:"active_command_id,omitempty"`
+	Summary         string `json:"summary,omitempty"`
+}
+
+// edgeCommandRequestBody is the JSON body for POST /edges/{id}/commands and
+// POST /fleet/commands. It carries the surface-neutral command intent only.
+type edgeCommandRequestBody struct {
+	Operation      string            `json:"operation"`
+	TargetSelector string            `json:"target_selector,omitempty"`
+	Parameters     map[string]string `json:"parameters,omitempty"`
+}
+
+type edgeCommandResponseView struct {
+	RequestID string `json:"request_id,omitempty"`
+	CommandID string `json:"command_id"`
+	EdgeID    string `json:"edge_id"`
+	Status    string `json:"status"`
+	Summary   string `json:"summary,omitempty"`
+	Error     string `json:"error,omitempty"`
+}
+
+type edgeOperationsResponse struct {
+	EdgeID     string                  `json:"edge_id"`
+	Operations []edgeCommandRecordView `json:"operations"`
+}
+
+type edgeCommandRecordView struct {
+	EdgeID         string                 `json:"edge_id"`
+	CommandID      string                 `json:"command_id"`
+	Operation      string                 `json:"operation,omitempty"`
+	TargetSelector string                 `json:"target_selector,omitempty"`
+	Status         string                 `json:"status"`
+	Summary        string                 `json:"summary,omitempty"`
+	Error          string                 `json:"error,omitempty"`
+	CreatedAt      time.Time              `json:"created_at"`
+	UpdatedAt      time.Time              `json:"updated_at"`
+	Events         []edgeCommandEventView `json:"events,omitempty"`
+}
+
+type edgeCommandEventView struct {
+	Phase      string    `json:"phase,omitempty"`
+	Summary    string    `json:"summary,omitempty"`
+	OccurredAt time.Time `json:"occurred_at,omitempty"`
+	ReceivedAt time.Time `json:"received_at"`
+}
+
+type fleetStatusResponse struct {
+	GeneratedAt time.Time       `json:"generated_at"`
+	Edges       []fleetEdgeView `json:"edges"`
+}
+
+type fleetEdgeView struct {
+	EdgeID       string                       `json:"edge_id"`
+	EdgeName     string                       `json:"edge_name,omitempty"`
+	Version      string                       `json:"version,omitempty"`
+	Protocol     string                       `json:"protocol,omitempty"`
+	Connected    bool                         `json:"connected"`
+	Health       string                       `json:"health"`
+	LastSeen     time.Time                    `json:"last_seen"`
+	NodeCount    int                          `json:"node_count"`
+	Capabilities []edgeCapabilitySummaryView  `json:"capabilities,omitempty"`
+	DomainAgents []edgeDomainAgentSummaryView `json:"domain_agents,omitempty"`
+	Error        string                       `json:"error,omitempty"`
+}
+
+type fleetCommandResponse struct {
+	Operation string                    `json:"operation"`
+	Results   []edgeCommandResponseView `json:"results"`
 }

 type edgeNodeSnapshotView struct {
@ -288,15 +372,21 @@ func newNodeConfigSummaryView(payload *iop.NodeConfigPayload) *nodeConfigSummary
 	}
 }

-func registerEdgeRegistryHandlers(mux *http.ServeMux, registry *wire.EdgeRegistry, requestStatus func(string, time.Duration) (*iop.EdgeStatusResponse, error)) {
+// edgeStatusRequester asks a connected Edge for its Edge-owned node snapshot.
+type edgeStatusRequester func(edgeID string, timeout time.Duration) (*iop.EdgeStatusResponse, error)
+
+// edgeCommandSender dispatches a command to a connected Edge and returns its
+// typed response.
+type edgeCommandSender func(edgeID, operation, targetSelector string, parameters map[string]string, timeout time.Duration) (*iop.EdgeCommandResponse, error)
+
+func registerEdgeRegistryHandlers(mux *http.ServeMux, registry *wire.EdgeRegistry, requestStatus edgeStatusRequester, sendCommand edgeCommandSender) {
 	mux.HandleFunc("/edges", func(w http.ResponseWriter, r *http.Request) {
 		if r.URL.Path != "/edges" {
 			http.NotFound(w, r)
 			return
 		}
 		if r.Method != http.MethodGet {
-			w.Header().Set("Allow", http.MethodGet)
-			http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
+			methodNotAllowed(w, http.MethodGet)
 			return
 		}
 		states := registry.Snapshots()
@ -307,52 +397,48 @@ func registerEdgeRegistryHandlers(mux *http.ServeMux, registry *wire.EdgeRegistr
 		writeJSON(w, http.StatusOK, edgeRegistryResponse{Edges: views})
 	})
 	mux.HandleFunc("/edges/", func(w http.ResponseWriter, r *http.Request) {
-		if r.Method != http.MethodGet {
-			w.Header().Set("Allow", http.MethodGet)
-			http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
-			return
-		}
 		edgePath := strings.TrimPrefix(r.URL.Path, "/edges/")
 		edgeID, rest, hasRest := strings.Cut(edgePath, "/")
 		if edgeID == "" {
 			http.NotFound(w, r)
 			return
 		}
-		if hasRest {
-			if rest != "events" && rest != "status" {
+		if !hasRest {
+			if r.Method != http.MethodGet {
+				methodNotAllowed(w, http.MethodGet)
+				return
+			}
+			state, ok := registry.Snapshot(edgeID)
+			if !ok {
 				http.NotFound(w, r)
 				return
 			}
+			writeJSON(w, http.StatusOK, edgeRegistryViewFromState(state))
+			return
+		}
+		switch rest {
+		case "status":
+			if r.Method != http.MethodGet {
+				methodNotAllowed(w, http.MethodGet)
+				return
+			}
 			if _, ok := registry.Snapshot(edgeID); !ok {
 				http.NotFound(w, r)
 				return
 			}
-			if rest == "status" {
-				resp, err := requestStatus(edgeID, 5*time.Second)
-				if err != nil {
-					writeJSON(w, http.StatusBadGateway, map[string]string{"error": err.Error()})
-					return
-				}
-				nodes := make([]edgeNodeSnapshotView, 0, len(resp.GetNodes()))
-				for _, n := range resp.GetNodes() {
-					nodes = append(nodes, edgeNodeSnapshotView{
-						NodeID:    n.GetNodeId(),
-						Alias:     n.GetAlias(),
-						Label:     n.GetLabel(),
-						Connected: n.GetConnected(),
-						Config:    newNodeConfigSummaryView(n.GetConfig()),
-					})
-				}
-				view := edgeStatusResponseView{
-					RequestID:            resp.GetRequestId(),
-					EdgeID:               resp.GetEdgeId(),
-					EdgeName:             resp.GetEdgeName(),
-					ObservedTimeUnixNano: resp.GetObservedTimeUnixNano(),
-					Nodes:                nodes,
-					Metadata:             resp.GetMetadata(),
-					Error:                resp.GetError(),
-				}
-				writeJSON(w, http.StatusOK, view)
+			resp, err := requestStatus(edgeID, 5*time.Second)
+			if err != nil {
+				writeJSON(w, http.StatusBadGateway, map[string]string{"error": err.Error()})
+				return
+			}
+			writeJSON(w, http.StatusOK, edgeStatusResponseViewFromProto(resp))
+		case "events":
+			if r.Method != http.MethodGet {
+				methodNotAllowed(w, http.MethodGet)
+				return
+			}
+			if _, ok := registry.Snapshot(edgeID); !ok {
+				http.NotFound(w, r)
 				return
 			}
 			records := registry.NodeEvents(edgeID, r.URL.Query().Get("node_id"))
@ -361,15 +447,111 @@ func registerEdgeRegistryHandlers(mux *http.ServeMux, registry *wire.EdgeRegistr
 				views = append(views, edgeNodeEventViewFromRecord(record))
 			}
 			writeJSON(w, http.StatusOK, edgeNodeEventsResponse{EdgeID: edgeID, Events: views})
-			return
-		}
-		state, ok := registry.Snapshot(edgeID)
-		if !ok {
+		case "operations":
+			if r.Method != http.MethodGet {
+				methodNotAllowed(w, http.MethodGet)
+				return
+			}
+			if _, ok := registry.Snapshot(edgeID); !ok {
+				http.NotFound(w, r)
+				return
+			}
+			records := registry.Commands(edgeID)
+			views := make([]edgeCommandRecordView, 0, len(records))
+			for _, record := range records {
+				views = append(views, edgeCommandRecordViewFromRecord(record))
+			}
+			writeJSON(w, http.StatusOK, edgeOperationsResponse{EdgeID: edgeID, Operations: views})
+		case "commands":
+			if r.Method != http.MethodPost {
+				methodNotAllowed(w, http.MethodPost)
+				return
+			}
+			handleEdgeCommandPost(w, r, edgeID, sendCommand)
+		default:
 			http.NotFound(w, r)
+		}
+	})
+}
+
+// handleEdgeCommandPost decodes a command request body and dispatches it to a
+// single target Edge. A not-connected Edge yields a 502 with the wire error.
+func handleEdgeCommandPost(w http.ResponseWriter, r *http.Request, edgeID string, sendCommand edgeCommandSender) {
+	var body edgeCommandRequestBody
+	if err := json.NewDecoder(r.Body).Decode(&body); err != nil {
+		writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid request body"})
+		return
+	}
+	if body.Operation == "" {
+		writeJSON(w, http.StatusBadRequest, map[string]string{"error": "operation is required"})
+		return
+	}
+	if sendCommand == nil {
+		writeJSON(w, http.StatusServiceUnavailable, map[string]string{"error": "command dispatch is not configured"})
+		return
+	}
+	resp, err := sendCommand(edgeID, body.Operation, body.TargetSelector, body.Parameters, 10*time.Second)
+	if err != nil {
+		writeJSON(w, http.StatusBadGateway, map[string]string{"error": err.Error()})
+		return
+	}
+	writeJSON(w, http.StatusAccepted, edgeCommandResponseViewFromProto(resp))
+}
+
+func registerFleetHandlers(mux *http.ServeMux, registry *wire.EdgeRegistry, requestStatus edgeStatusRequester, sendCommand edgeCommandSender) {
+	mux.HandleFunc("/fleet/status", func(w http.ResponseWriter, r *http.Request) {
+		if r.Method != http.MethodGet {
+			methodNotAllowed(w, http.MethodGet)
 			return
 		}
-		writeJSON(w, http.StatusOK, edgeRegistryViewFromState(state))
+		states := registry.Snapshots()
+		views := make([]fleetEdgeView, 0, len(states))
+		for _, state := range states {
+			views = append(views, fleetEdgeViewFromState(state, requestStatus))
+		}
+		writeJSON(w, http.StatusOK, fleetStatusResponse{GeneratedAt: time.Now().UTC(), Edges: views})
 	})
+	mux.HandleFunc("/fleet/commands", func(w http.ResponseWriter, r *http.Request) {
+		if r.Method != http.MethodPost {
+			methodNotAllowed(w, http.MethodPost)
+			return
+		}
+		var body edgeCommandRequestBody
+		if err := json.NewDecoder(r.Body).Decode(&body); err != nil {
+			writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid request body"})
+			return
+		}
+		if body.Operation == "" {
+			writeJSON(w, http.StatusBadRequest, map[string]string{"error": "operation is required"})
+			return
+		}
+		if sendCommand == nil {
+			writeJSON(w, http.StatusServiceUnavailable, map[string]string{"error": "command dispatch is not configured"})
+			return
+		}
+		results := make([]edgeCommandResponseView, 0)
+		for _, state := range registry.Snapshots() {
+			if !state.Connected {
+				continue
+			}
+			resp, err := sendCommand(state.EdgeID, body.Operation, body.TargetSelector, body.Parameters, 10*time.Second)
+			if err != nil {
+				results = append(results, edgeCommandResponseView{
+					EdgeID: state.EdgeID,
+					Status: "error",
+					Error:  err.Error(),
+				})
+				continue
+			}
+			results = append(results, edgeCommandResponseViewFromProto(resp))
+		}
+		writeJSON(w, http.StatusAccepted, fleetCommandResponse{Operation: body.Operation, Results: results})
+	})
+}
+
+func methodNotAllowed(w http.ResponseWriter, allow ...string) {
+	w.Header().Set("Allow", strings.Join(allow, ", "))
+	http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
 }

 func edgeNodeEventViewFromRecord(record wire.EdgeNodeEventRecord) edgeNodeEventView {
@ -411,6 +593,135 @@ func edgeRegistryViewFromState(state wire.EdgeConnectionState) edgeRegistryView
 	}
 }

+func edgeStatusResponseViewFromProto(resp *iop.EdgeStatusResponse) edgeStatusResponseView {
+	nodes := make([]edgeNodeSnapshotView, 0, len(resp.GetNodes()))
+	for _, n := range resp.GetNodes() {
+		nodes = append(nodes, edgeNodeSnapshotView{
+			NodeID:    n.GetNodeId(),
+			Alias:     n.GetAlias(),
+			Label:     n.GetLabel(),
+			Connected: n.GetConnected(),
+			Config:    newNodeConfigSummaryView(n.GetConfig()),
+		})
+	}
+	return edgeStatusResponseView{
+		RequestID:            resp.GetRequestId(),
+		EdgeID:               resp.GetEdgeId(),
+		EdgeName:             resp.GetEdgeName(),
+		ObservedTimeUnixNano: resp.GetObservedTimeUnixNano(),
+		Nodes:                nodes,
+		Capabilities:         edgeCapabilityViews(resp.GetCapabilities()),
+		DomainAgents:         edgeDomainAgentViews(resp.GetDomainAgents()),
+		Metadata:             resp.GetMetadata(),
+		Error:                resp.GetError(),
+	}
+}
+
+func edgeCapabilityViews(caps []*iop.EdgeCapabilitySummary) []edgeCapabilitySummaryView {
+	if len(caps) == 0 {
+		return nil
+	}
+	views := make([]edgeCapabilitySummaryView, 0, len(caps))
+	for _, c := range caps {
+		views = append(views, edgeCapabilitySummaryView{
+			Kind:      c.GetKind(),
+			Available: c.GetAvailable(),
+			Status:    c.GetStatus(),
+			Summary:   c.GetSummary(),
+		})
+	}
+	return views
+}
+
+func edgeDomainAgentViews(agents []*iop.EdgeDomainAgentSummary) []edgeDomainAgentSummaryView {
+	if len(agents) == 0 {
+		return nil
+	}
+	views := make([]edgeDomainAgentSummaryView, 0, len(agents))
+	for _, a := range agents {
+		views = append(views, edgeDomainAgentSummaryView{
+			AgentKind:       a.GetAgentKind(),
+			Available:       a.GetAvailable(),
+			LifecycleState:  a.GetLifecycleState(),
+			ActiveCommandID: a.GetActiveCommandId(),
+			Summary:         a.GetSummary(),
+		})
+	}
+	return views
+}
+
+func edgeCommandResponseViewFromProto(resp *iop.EdgeCommandResponse) edgeCommandResponseView {
+	return edgeCommandResponseView{
+		RequestID: resp.GetRequestId(),
+		CommandID: resp.GetCommandId(),
+		EdgeID:    resp.GetEdgeId(),
+		Status:    resp.GetStatus(),
+		Summary:   resp.GetSummary(),
+		Error:     resp.GetError(),
+	}
+}
+
+func edgeCommandRecordViewFromRecord(record wire.EdgeCommandRecord) edgeCommandRecordView {
+	view := edgeCommandRecordView{
+		EdgeID:         record.EdgeID,
+		CommandID:      record.CommandID,
+		Operation:      record.Operation,
+		TargetSelector: record.TargetSelector,
+		Status:         record.Status,
+		Summary:        record.Summary,
+		Error:          record.Error,
+		CreatedAt:      record.CreatedAt,
+		UpdatedAt:      record.UpdatedAt,
+	}
+	if len(record.Events) > 0 {
+		view.Events = make([]edgeCommandEventView, 0, len(record.Events))
+		for _, e := range record.Events {
+			view.Events = append(view.Events, edgeCommandEventView{
+				Phase:      e.Phase,
+				Summary:    e.Summary,
+				OccurredAt: e.OccurredAt,
+				ReceivedAt: e.ReceivedAt,
+			})
+		}
+	}
+	return view
+}
+
+func fleetEdgeViewFromState(state wire.EdgeConnectionState, requestStatus edgeStatusRequester) fleetEdgeView {
+	view := fleetEdgeView{
+		EdgeID:    state.EdgeID,
+		EdgeName:  state.EdgeName,
+		Version:   state.Version,
+		Protocol:  state.Protocol,
+		Connected: state.Connected,
+		LastSeen:  state.LastSeen,
+	}
+	if !state.Connected {
+		view.Health = "offline"
+		return view
+	}
+	if requestStatus == nil {
+		view.Health = "online"
+		return view
+	}
+	resp, err := requestStatus(state.EdgeID, 5*time.Second)
+	if err != nil {
+		view.Health = "degraded"
+		view.Error = err.Error()
+		return view
+	}
+	view.NodeCount = len(resp.GetNodes())
+	view.Capabilities = edgeCapabilityViews(resp.GetCapabilities())
+	view.DomainAgents = edgeDomainAgentViews(resp.GetDomainAgents())
+	if resp.GetError() != "" {
+		view.Health = "degraded"
+		view.Error = resp.GetError()
+		return view
+	}
+	view.Health = "online"
+	return view
+}
+
 func writeJSON(w http.ResponseWriter, status int, v any) {
 	w.Header().Set("Content-Type", "application/json")
 	w.WriteHeader(status)
--- a/apps/control-plane/cmd/control-plane/main_test.go
+++ b/apps/control-plane/cmd/control-plane/main_test.go
@ -11,6 +11,7 @@ import (
 	"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"
@ -256,7 +257,7 @@ func TestEdgeRegistryHTTPHandlersListAndGetEdge(t *testing.T) {
 	}, at.Add(time.Second))

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

 	listResp := httptest.NewRecorder()
 	mux.ServeHTTP(listResp, httptest.NewRequest(http.MethodGet, "/edges", nil))
@ -334,7 +335,7 @@ func TestEdgeRegistryHTTPHandlersListNodeEvents(t *testing.T) {
 	}, receivedAt)

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

 	resp := httptest.NewRecorder()
 	mux.ServeHTTP(resp, httptest.NewRequest(http.MethodGet, "/edges/edge-a/events", nil))
@ -397,7 +398,7 @@ func TestEdgeRegistryHTTPHandlersListNodeEvents(t *testing.T) {
 func TestEdgeRegistryHTTPHandlersRejectUnsupportedCases(t *testing.T) {
 	registry := wire.NewEdgeRegistry()
 	mux := http.NewServeMux()
-	registerEdgeRegistryHandlers(mux, registry, nil)
+	registerEdgeRegistryHandlers(mux, registry, nil, nil)

 	postResp := httptest.NewRecorder()
 	mux.ServeHTTP(postResp, httptest.NewRequest(http.MethodPost, "/edges", nil))
@ -463,7 +464,7 @@ func TestEdgeRegistryHTTPHandlersGetLiveStatus(t *testing.T) {
 			}, nil
 		}

-		registerEdgeRegistryHandlers(mux, registry, requestStatus)
+		registerEdgeRegistryHandlers(mux, registry, requestStatus, nil)

 		resp := httptest.NewRecorder()
 		mux.ServeHTTP(resp, httptest.NewRequest(http.MethodGet, "/edges/edge-a/status", nil))
@ -511,7 +512,7 @@ func TestEdgeRegistryHTTPHandlersGetLiveStatus(t *testing.T) {
 		requestStatus := func(edgeID string, timeout time.Duration) (*iop.EdgeStatusResponse, error) {
 			return nil, fmt.Errorf("should not be called")
 		}
-		registerEdgeRegistryHandlers(mux, registry, requestStatus)
+		registerEdgeRegistryHandlers(mux, registry, requestStatus, nil)

 		resp := httptest.NewRecorder()
 		mux.ServeHTTP(resp, httptest.NewRequest(http.MethodGet, "/edges/missing-edge/status", nil))
@ -526,7 +527,7 @@ func TestEdgeRegistryHTTPHandlersGetLiveStatus(t *testing.T) {
 		requestStatus := func(edgeID string, timeout time.Duration) (*iop.EdgeStatusResponse, error) {
 			return nil, fmt.Errorf("upstream timeout")
 		}
-		registerEdgeRegistryHandlers(mux, registry, requestStatus)
+		registerEdgeRegistryHandlers(mux, registry, requestStatus, nil)

 		resp := httptest.NewRecorder()
 		mux.ServeHTTP(resp, httptest.NewRequest(http.MethodGet, "/edges/edge-a/status", nil))
@ -544,3 +545,232 @@ func TestEdgeRegistryHTTPHandlersGetLiveStatus(t *testing.T) {
 		}
 	})
 }
+
+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.RecordCommandRequest("edge-a", &iop.EdgeCommandRequest{CommandId: "cmd-1", Operation: "restart-node", TargetSelector: "node-1"}, at)
+	registry.RecordCommandEvent("edge-a", &iop.EdgeCommandEvent{CommandId: "cmd-1", Phase: "running", Summary: "executing"}, at.Add(time.Second))
+	registry.RecordCommandResult("edge-a", &iop.EdgeCommandResponse{CommandId: "cmd-1", Status: "ok", Summary: "done"}, at.Add(2*time.Second))
+
+	mux := http.NewServeMux()
+	registerEdgeRegistryHandlers(mux, registry, nil, nil)
+
+	resp := httptest.NewRecorder()
+	mux.ServeHTTP(resp, httptest.NewRequest(http.MethodGet, "/edges/edge-a/operations", nil))
+	if resp.Code != http.StatusOK {
+		t.Fatalf("GET operations status=%d body=%s", resp.Code, resp.Body.String())
+	}
+	var got edgeOperationsResponse
+	if err := json.Unmarshal(resp.Body.Bytes(), &got); err != nil {
+		t.Fatalf("decode operations: %v", err)
+	}
+	if got.EdgeID != "edge-a" || len(got.Operations) != 1 {
+		t.Fatalf("unexpected operations response: %+v", got)
+	}
+	op := got.Operations[0]
+	if op.CommandID != "cmd-1" || op.Operation != "restart-node" || op.Status != "ok" || op.Summary != "done" {
+		t.Fatalf("unexpected operation record: %+v", op)
+	}
+	if len(op.Events) != 1 || op.Events[0].Phase != "running" {
+		t.Fatalf("unexpected operation events: %+v", op.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-degraded", EdgeName: "Degraded"}, at)
+	registry.MarkConnected(&iop.EdgeHelloRequest{EdgeId: "edge-offline", EdgeName: "Offline"}, at)
+	registry.MarkDisconnected("edge-offline", 3, 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"},
+				},
+			}, 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) != 3 {
+		t.Fatalf("expected 3 fleet edges, got %d", len(got.Edges))
+	}
+	health := map[string]fleetEdgeView{}
+	for _, e := range got.Edges {
+		health[e.EdgeID] = e
+	}
+	if e := health["edge-online"]; e.Health != "online" || e.NodeCount != 1 || len(e.Capabilities) != 1 {
+		t.Fatalf("unexpected online edge view: %+v", e)
+	}
+	if e := health["edge-degraded"]; e.Health != "degraded" || e.Error == "" {
+		t.Fatalf("unexpected degraded edge view: %+v", e)
+	}
+	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)
+		}
+	}
+}
--- a/apps/control-plane/internal/wire/edge_registry.go
+++ b/apps/control-plane/internal/wire/edge_registry.go
@ -11,6 +11,8 @@ import (
 )

 const recentNodeEventCapacity = 128
+const recentCommandCapacity = 128
+const commandEventCapacity = 64

 // EdgeConnectionState captures the Control Plane's internal view of a single
 // Edge wire connection.
@ -38,12 +40,39 @@ type EdgeNodeEventRecord struct {
 	ReceivedAt time.Time
 }

+// EdgeCommandRecord is the Control Plane's bounded audit view of one command
+// dispatched to an Edge: the request summary, the latest result, and a bounded
+// list of relayed lifecycle events. It is a control/audit view; durable command
+// history is intentionally outside this in-memory baseline.
+type EdgeCommandRecord struct {
+	EdgeID         string
+	CommandID      string
+	Operation      string
+	TargetSelector string
+	Status         string
+	Summary        string
+	Error          string
+	CreatedAt      time.Time
+	UpdatedAt      time.Time
+	Events         []EdgeCommandEventRecord
+}
+
+// EdgeCommandEventRecord is the Control Plane's copied view of one command
+// lifecycle event relayed from an Edge during command execution.
+type EdgeCommandEventRecord struct {
+	Phase      string
+	Summary    string
+	OccurredAt time.Time
+	ReceivedAt time.Time
+}
+
 // EdgeRegistry stores Edge connection state keyed by edge_id behind a mutex.
 type EdgeRegistry struct {
 	mu               sync.Mutex
 	edges            map[string]*EdgeConnectionState
 	nextToken        uint64
 	recentNodeEvents []EdgeNodeEventRecord
+	commands         []*EdgeCommandRecord
 }

 // NewEdgeRegistry returns an empty Edge connection registry.
@ -194,6 +223,172 @@ func (r *EdgeRegistry) NodeEvents(edgeID, nodeID string) []EdgeNodeEventRecord {
 	return out
 }

+// RecordCommandRequest stores or refreshes a pending command audit record for a
+// command the Control Plane is dispatching to an Edge. It is bounded; the oldest
+// record is evicted past capacity.
+func (r *EdgeRegistry) RecordCommandRequest(edgeID string, req *iop.EdgeCommandRequest, at time.Time) {
+	if edgeID == "" || req == nil || req.GetCommandId() == "" {
+		return
+	}
+	if at.IsZero() {
+		at = time.Now()
+	}
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	if rec := r.findCommandLocked(edgeID, req.GetCommandId()); rec != nil {
+		rec.Operation = req.GetOperation()
+		rec.TargetSelector = req.GetTargetSelector()
+		rec.UpdatedAt = at
+		return
+	}
+	rec := r.appendCommandLocked(edgeID, req.GetCommandId(), at)
+	rec.Operation = req.GetOperation()
+	rec.TargetSelector = req.GetTargetSelector()
+	rec.Status = "pending"
+}
+
+// RecordCommandResult updates the audit record for a command with the Edge's
+// response status/summary. A missing record is created so a result is never
+// silently dropped.
+func (r *EdgeRegistry) RecordCommandResult(edgeID string, resp *iop.EdgeCommandResponse, at time.Time) {
+	if edgeID == "" || resp == nil || resp.GetCommandId() == "" {
+		return
+	}
+	if at.IsZero() {
+		at = time.Now()
+	}
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	rec := r.findCommandLocked(edgeID, resp.GetCommandId())
+	if rec == nil {
+		rec = r.appendCommandLocked(edgeID, resp.GetCommandId(), at)
+	}
+	status := resp.GetStatus()
+	if status == "" {
+		status = "ok"
+	}
+	rec.Status = status
+	rec.Summary = resp.GetSummary()
+	rec.Error = resp.GetError()
+	rec.UpdatedAt = at
+}
+
+// RecordCommandEventFromConnection appends a bounded command lifecycle event to
+// the matching command audit record, but only when token still identifies the
+// current connection for edgeID. It returns false for stale superseded
+// connections.
+func (r *EdgeRegistry) RecordCommandEventFromConnection(edgeID string, token uint64, event *iop.EdgeCommandEvent, receivedAt time.Time) bool {
+	return r.recordCommandEvent(edgeID, token, true, event, receivedAt)
+}
+
+// RecordCommandEvent stores a command lifecycle event without a connection
+// token. It is intended for tests and non-connection seed paths.
+func (r *EdgeRegistry) RecordCommandEvent(edgeID string, event *iop.EdgeCommandEvent, receivedAt time.Time) {
+	r.recordCommandEvent(edgeID, 0, false, event, receivedAt)
+}
+
+func (r *EdgeRegistry) recordCommandEvent(edgeID string, token uint64, requireToken bool, event *iop.EdgeCommandEvent, receivedAt time.Time) bool {
+	if edgeID == "" || event == nil || event.GetCommandId() == "" {
+		return false
+	}
+	if receivedAt.IsZero() {
+		receivedAt = time.Now()
+	}
+	occurredAt := receivedAt
+	if event.GetOccurredAt() != 0 {
+		occurredAt = time.Unix(0, event.GetOccurredAt())
+	}
+
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	if requireToken {
+		state, ok := r.edges[edgeID]
+		if !ok || state.token != token {
+			return false
+		}
+	}
+	rec := r.findCommandLocked(edgeID, event.GetCommandId())
+	if rec == nil {
+		rec = r.appendCommandLocked(edgeID, event.GetCommandId(), receivedAt)
+		rec.Status = "running"
+	}
+	if len(rec.Events) >= commandEventCapacity {
+		rec.Events = append(rec.Events[:0], rec.Events[1:]...)
+	}
+	rec.Events = append(rec.Events, EdgeCommandEventRecord{
+		Phase:      event.GetPhase(),
+		Summary:    event.GetSummary(),
+		OccurredAt: occurredAt,
+		ReceivedAt: receivedAt,
+	})
+	rec.UpdatedAt = receivedAt
+	return true
+}
+
+// Commands returns copied command audit records for an Edge, oldest first.
+func (r *EdgeRegistry) Commands(edgeID string) []EdgeCommandRecord {
+	if edgeID == "" {
+		return nil
+	}
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	out := make([]EdgeCommandRecord, 0, len(r.commands))
+	for _, rec := range r.commands {
+		if rec.EdgeID != edgeID {
+			continue
+		}
+		out = append(out, cloneEdgeCommandRecord(rec))
+	}
+	if len(out) == 0 {
+		return nil
+	}
+	return out
+}
+
+// Command returns a single copied command audit record.
+func (r *EdgeRegistry) Command(edgeID, commandID string) (EdgeCommandRecord, bool) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	rec := r.findCommandLocked(edgeID, commandID)
+	if rec == nil {
+		return EdgeCommandRecord{}, false
+	}
+	return cloneEdgeCommandRecord(rec), true
+}
+
+// findCommandLocked returns the stored command record for an edge_id/command_id
+// pair. Callers must hold r.mu.
+func (r *EdgeRegistry) findCommandLocked(edgeID, commandID string) *EdgeCommandRecord {
+	for _, rec := range r.commands {
+		if rec.EdgeID == edgeID && rec.CommandID == commandID {
+			return rec
+		}
+	}
+	return nil
+}
+
+// appendCommandLocked appends a new bounded command record. Callers must hold
+// r.mu.
+func (r *EdgeRegistry) appendCommandLocked(edgeID, commandID string, at time.Time) *EdgeCommandRecord {
+	if len(r.commands) >= recentCommandCapacity {
+		r.commands = append(r.commands[:0], r.commands[1:]...)
+	}
+	rec := &EdgeCommandRecord{
+		EdgeID:    edgeID,
+		CommandID: commandID,
+		CreatedAt: at,
+		UpdatedAt: at,
+	}
+	r.commands = append(r.commands, rec)
+	return rec
+}
+
+func cloneEdgeCommandRecord(rec *EdgeCommandRecord) EdgeCommandRecord {
+	clone := *rec
+	clone.Events = append([]EdgeCommandEventRecord(nil), rec.Events...)
+	return clone
+}
+
 // Len returns the number of tracked Edge connections.
 func (r *EdgeRegistry) Len() int {
 	r.mu.Lock()
--- a/apps/control-plane/internal/wire/edge_server.go
+++ b/apps/control-plane/internal/wire/edge_server.go
@ -83,6 +83,17 @@ func NewEdgeServer(listenAddr string, logger *zap.Logger) (*EdgeServer, error) {
 			registry.RecordNodeEventFromConnection(id, tok, event, time.Now())
 		})

+		proto_socket.AddListenerTyped[*iop.EdgeCommandEvent](&client.Communicator, func(event *iop.EdgeCommandEvent) {
+			idMu.Lock()
+			id := edgeID
+			tok := token
+			idMu.Unlock()
+			if id == "" {
+				return
+			}
+			registry.RecordCommandEventFromConnection(id, tok, event, time.Now())
+		})
+
 		proto_socket.AddRequestListenerTyped(&client.Communicator, func(req *iop.EdgeHelloRequest) (*iop.EdgeHelloResponse, error) {
 			now := time.Now()
 			if req.GetEdgeId() == "" {
@ -205,6 +216,52 @@ func (s *EdgeServer) RequestStatus(edgeID string, timeout time.Duration) (*iop.E
 	)
 }

+// SendCommand sends a command request to a connected Edge over the wire, records
+// the request and result in the audit registry, and returns the Edge's typed
+// response. It routes only to the target Edge's active connection and fails when
+// that Edge is not currently connected. Command lifecycle events the Edge relays
+// during execution are recorded asynchronously by the connection's event
+// listener. It observes and controls through the Edge; it does not reach Node
+// directly.
+func (s *EdgeServer) SendCommand(edgeID, operation, targetSelector string, parameters map[string]string, timeout time.Duration) (*iop.EdgeCommandResponse, error) {
+	client, ok := s.activeClient(edgeID)
+	if !ok {
+		return nil, fmt.Errorf("edge %q is not connected", edgeID)
+	}
+	now := time.Now()
+	req := &iop.EdgeCommandRequest{
+		RequestId:      fmt.Sprintf("edge-command-%d", now.UnixNano()),
+		CommandId:      fmt.Sprintf("cmd-%d", now.UnixNano()),
+		Operation:      operation,
+		TargetSelector: targetSelector,
+		Parameters:     parameters,
+	}
+	s.registry.RecordCommandRequest(edgeID, req, now)
+
+	resp, err := proto_socket.SendRequestTyped[*iop.EdgeCommandRequest, *iop.EdgeCommandResponse](
+		&client.Communicator,
+		req,
+		timeout,
+	)
+	if err != nil {
+		s.registry.RecordCommandResult(edgeID, &iop.EdgeCommandResponse{
+			CommandId: req.GetCommandId(),
+			EdgeId:    edgeID,
+			Status:    "error",
+			Error:     err.Error(),
+		}, time.Now())
+		return nil, err
+	}
+	if resp.GetCommandId() == "" {
+		resp.CommandId = req.GetCommandId()
+	}
+	if resp.GetEdgeId() == "" {
+		resp.EdgeId = edgeID
+	}
+	s.registry.RecordCommandResult(edgeID, resp, time.Now())
+	return resp, nil
+}
+
 // Host returns the host the server is listening on.
 func (s *EdgeServer) Host() string { return s.host }

--- a/apps/control-plane/internal/wire/edge_server_test.go
+++ b/apps/control-plane/internal/wire/edge_server_test.go
@ -3,6 +3,7 @@ package wire
 import (
 	"context"
 	"fmt"
+	"sync"
 	"testing"
 	"time"

@ -404,6 +405,182 @@ func TestEdgeServerRecordsNodeLifecycleEventFromEdge(t *testing.T) {
 	}
 }

+type commandEdgeRecorder struct {
+	mu       sync.Mutex
+	received []string
+}
+
+func (r *commandEdgeRecorder) operations() []string {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	return append([]string(nil), r.received...)
+}
+
+// dialCommandEdge dials the server as an Edge that answers command requests with
+// an "ok" response and records the operations it was asked to run.
+func dialCommandEdge(t *testing.T, ctx context.Context, port int, edgeID string) *commandEdgeRecorder {
+	t.Helper()
+	client := dialEdge(t, ctx, port)
+	t.Cleanup(func() { _ = client.Close() })
+	rec := &commandEdgeRecorder{}
+	proto_socket.AddRequestListenerTyped(&client.Communicator, func(req *iop.EdgeCommandRequest) (*iop.EdgeCommandResponse, error) {
+		rec.mu.Lock()
+		rec.received = append(rec.received, req.GetOperation())
+		rec.mu.Unlock()
+		return &iop.EdgeCommandResponse{
+			RequestId: req.GetRequestId(),
+			CommandId: req.GetCommandId(),
+			EdgeId:    edgeID,
+			Status:    "ok",
+			Summary:   "applied " + req.GetOperation(),
+		}, nil
+	})
+	sendEdgeHello(t, client, edgeID)
+	return rec
+}
+
+func TestEdgeServerSendsCommandToTargetEdgeOnly(t *testing.T) {
+	server, port := startEdgeServer(t)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	recA := dialCommandEdge(t, ctx, port, "edge-a")
+	recB := dialCommandEdge(t, ctx, port, "edge-b")
+
+	resp, err := server.SendCommand("edge-a", "restart-node", "node-1", map[string]string{"force": "true"}, 2*time.Second)
+	if err != nil {
+		t.Fatalf("SendCommand: %v", err)
+	}
+	if resp.GetStatus() != "ok" || resp.GetEdgeId() != "edge-a" {
+		t.Fatalf("unexpected response: %+v", resp)
+	}
+	if resp.GetCommandId() == "" {
+		t.Error("expected a non-empty command_id correlation")
+	}
+
+	if got := recA.operations(); len(got) != 1 || got[0] != "restart-node" {
+		t.Fatalf("edge-a received: %+v", got)
+	}
+	if got := recB.operations(); len(got) != 0 {
+		t.Fatalf("edge-b should not receive a targeted command, got: %+v", got)
+	}
+
+	cmds := server.Registry().Commands("edge-a")
+	if len(cmds) != 1 {
+		t.Fatalf("edge-a audit: got %d want 1", len(cmds))
+	}
+	if cmds[0].Operation != "restart-node" || cmds[0].Status != "ok" {
+		t.Fatalf("unexpected audit record: %+v", cmds[0])
+	}
+	if got := server.Registry().Commands("edge-b"); len(got) != 0 {
+		t.Fatalf("edge-b audit should be empty, got %+v", got)
+	}
+}
+
+func TestEdgeServerSendCommandFailsWhenEdgeNotConnected(t *testing.T) {
+	server, _ := startEdgeServer(t)
+	if _, err := server.SendCommand("missing-edge", "restart-node", "", nil, time.Second); err == nil {
+		t.Fatal("expected error sending command to a non-connected edge")
+	}
+	if got := server.Registry().Commands("missing-edge"); len(got) != 0 {
+		t.Fatalf("expected no audit record for a never-connected edge, got %+v", got)
+	}
+}
+
+func TestEdgeServerRecordsCommandLifecycleEventsFromEdge(t *testing.T) {
+	server, port := startEdgeServer(t)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	client := dialEdge(t, ctx, port)
+	defer client.Close()
+	const edgeID = "edge-dgx-group"
+
+	proto_socket.AddRequestListenerTyped(&client.Communicator, func(req *iop.EdgeCommandRequest) (*iop.EdgeCommandResponse, error) {
+		_ = client.Send(&iop.EdgeCommandEvent{CommandId: req.GetCommandId(), EdgeId: edgeID, Phase: "accepted", Summary: "queued"})
+		_ = client.Send(&iop.EdgeCommandEvent{CommandId: req.GetCommandId(), EdgeId: edgeID, Phase: "running", Summary: "executing"})
+		return &iop.EdgeCommandResponse{
+			RequestId: req.GetRequestId(),
+			CommandId: req.GetCommandId(),
+			EdgeId:    edgeID,
+			Status:    "ok",
+			Summary:   "done",
+		}, nil
+	})
+
+	sendEdgeHello(t, client, edgeID)
+
+	resp, err := server.SendCommand(edgeID, "drain", "", nil, 2*time.Second)
+	if err != nil {
+		t.Fatalf("SendCommand: %v", err)
+	}
+
+	var rec EdgeCommandRecord
+	deadline := time.Now().Add(2 * time.Second)
+	for {
+		got, ok := server.Registry().Command(edgeID, resp.GetCommandId())
+		if ok && len(got.Events) == 2 {
+			rec = got
+			break
+		}
+		if time.Now().After(deadline) {
+			t.Fatalf("command events not recorded in time; record=%+v", got)
+		}
+		time.Sleep(20 * time.Millisecond)
+	}
+	if rec.Status != "ok" || rec.Summary != "done" {
+		t.Fatalf("unexpected final command record: %+v", rec)
+	}
+	if rec.Events[0].Phase != "accepted" || rec.Events[1].Phase != "running" {
+		t.Fatalf("unexpected event phases: %+v", rec.Events)
+	}
+}
+
+func TestEdgeRegistryCommandsAreBoundedAndCopied(t *testing.T) {
+	registry := NewEdgeRegistry()
+	at := time.Unix(1780142500, 0).UTC()
+
+	for i := 0; i < recentCommandCapacity+2; i++ {
+		cmdID := fmt.Sprintf("cmd-%03d", i)
+		registry.RecordCommandRequest("edge-a", &iop.EdgeCommandRequest{CommandId: cmdID, Operation: "op"}, at.Add(time.Duration(i)*time.Second))
+		registry.RecordCommandResult("edge-a", &iop.EdgeCommandResponse{CommandId: cmdID, Status: "ok", Summary: "done"}, at.Add(time.Duration(i)*time.Second).Add(time.Millisecond))
+	}
+
+	cmds := registry.Commands("edge-a")
+	if len(cmds) != recentCommandCapacity {
+		t.Fatalf("commands len: got %d want %d", len(cmds), recentCommandCapacity)
+	}
+	if cmds[0].CommandID != "cmd-002" {
+		t.Fatalf("expected oldest retained cmd-002, got %q", cmds[0].CommandID)
+	}
+	if cmds[0].Status != "ok" {
+		t.Fatalf("expected result status recorded, got %q", cmds[0].Status)
+	}
+
+	registry.RecordCommandRequest("edge-b", &iop.EdgeCommandRequest{CommandId: "cmd-b", Operation: "drain"}, at)
+	for i := 0; i < commandEventCapacity+3; i++ {
+		registry.RecordCommandEvent("edge-b", &iop.EdgeCommandEvent{CommandId: "cmd-b", Phase: fmt.Sprintf("phase-%03d", i)}, at.Add(time.Duration(i)*time.Nanosecond))
+	}
+	rec, ok := registry.Command("edge-b", "cmd-b")
+	if !ok {
+		t.Fatal("edge-b cmd-b missing")
+	}
+	if len(rec.Events) != commandEventCapacity {
+		t.Fatalf("events len: got %d want %d", len(rec.Events), commandEventCapacity)
+	}
+	if rec.Events[0].Phase != "phase-003" {
+		t.Fatalf("expected oldest retained phase-003, got %q", rec.Events[0].Phase)
+	}
+
+	rec.Events[0].Phase = "mutated"
+	again, _ := registry.Command("edge-b", "cmd-b")
+	if again.Events[0].Phase == "mutated" {
+		t.Fatal("Command returned mutable registry storage")
+	}
+}
+
 func TestEdgeServerRequestStatusFailsWhenEdgeNotConnected(t *testing.T) {
 	server, _ := startEdgeServer(t)
 	if _, err := server.RequestStatus("missing-edge", time.Second); err == nil {