Add instance agent control plane and runtime console

2026-04-04 22:57:41 +08:00
parent 58936726ed
commit 4ebb5f46d1
33 changed files with 5630 additions and 855 deletions
@@ -0,0 +1,337 @@
+package services
+
+import (
+	"encoding/json"
+	"fmt"
+	"strings"
+	"time"
+
+	"clawreef/internal/models"
+	"clawreef/internal/repository"
+)
+
+const (
+	InstanceCommandTypeStartOpenClaw       = "start_openclaw"
+	InstanceCommandTypeStopOpenClaw        = "stop_openclaw"
+	InstanceCommandTypeRestartOpenClaw     = "restart_openclaw"
+	InstanceCommandTypeCollectSystemInfo   = "collect_system_info"
+	InstanceCommandTypeApplyConfigRevision = "apply_config_revision"
+	InstanceCommandTypeReloadConfig        = "reload_config"
+	InstanceCommandTypeHealthCheck         = "health_check"
+	instanceCommandStatusPending           = "pending"
+	instanceCommandStatusDispatched        = "dispatched"
+	instanceCommandStatusRunning           = "running"
+	instanceCommandStatusSucceeded         = "succeeded"
+	instanceCommandStatusFailed            = "failed"
+)
+
+type CreateInstanceCommandRequest struct {
+	CommandType    string                 `json:"command_type"`
+	Payload        map[string]interface{} `json:"payload,omitempty"`
+	IdempotencyKey string                 `json:"idempotency_key"`
+	TimeoutSeconds int                    `json:"timeout_seconds"`
+}
+
+type InstanceCommandPayload struct {
+	ID             int                    `json:"id"`
+	CommandType    string                 `json:"command_type"`
+	Payload        map[string]interface{} `json:"payload,omitempty"`
+	Status         string                 `json:"status"`
+	IdempotencyKey string                 `json:"idempotency_key"`
+	IssuedBy       *int                   `json:"issued_by,omitempty"`
+	IssuedAt       time.Time              `json:"issued_at"`
+	DispatchedAt   *time.Time             `json:"dispatched_at,omitempty"`
+	StartedAt      *time.Time             `json:"started_at,omitempty"`
+	FinishedAt     *time.Time             `json:"finished_at,omitempty"`
+	TimeoutSeconds int                    `json:"timeout_seconds"`
+	Result         map[string]interface{} `json:"result,omitempty"`
+	ErrorMessage   *string                `json:"error_message,omitempty"`
+}
+
+type AgentCommandEnvelope struct {
+	ID             int                    `json:"id"`
+	CommandType    string                 `json:"command_type"`
+	Payload        map[string]interface{} `json:"payload,omitempty"`
+	IssuedAt       time.Time              `json:"issued_at"`
+	TimeoutSeconds int                    `json:"timeout_seconds"`
+	IdempotencyKey string                 `json:"idempotency_key"`
+}
+
+type AgentCommandFinishRequest struct {
+	AgentID      string                 `json:"agent_id" binding:"required"`
+	Status       string                 `json:"status" binding:"required"`
+	FinishedAt   *time.Time             `json:"finished_at,omitempty"`
+	Result       map[string]interface{} `json:"result,omitempty"`
+	ErrorMessage string                 `json:"error_message"`
+}
+
+type InstanceCommandService interface {
+	Create(instanceID int, issuedBy *int, req CreateInstanceCommandRequest) (*InstanceCommandPayload, error)
+	GetNextForAgent(session *AgentSession) (*AgentCommandEnvelope, error)
+	MarkStarted(session *AgentSession, commandID int, startedAt *time.Time) error
+	MarkFinished(session *AgentSession, commandID int, req AgentCommandFinishRequest) error
+	ListByInstanceID(instanceID int, limit int) ([]InstanceCommandPayload, error)
+}
+
+type instanceCommandService struct {
+	commandRepo      repository.InstanceCommandRepository
+	runtimeRepo      repository.InstanceRuntimeStatusRepository
+	desiredStateRepo repository.InstanceDesiredStateRepository
+}
+
+func NewInstanceCommandService(commandRepo repository.InstanceCommandRepository, runtimeRepo repository.InstanceRuntimeStatusRepository, desiredStateRepo repository.InstanceDesiredStateRepository) InstanceCommandService {
+	return &instanceCommandService{
+		commandRepo:      commandRepo,
+		runtimeRepo:      runtimeRepo,
+		desiredStateRepo: desiredStateRepo,
+	}
+}
+
+func (s *instanceCommandService) Create(instanceID int, issuedBy *int, req CreateInstanceCommandRequest) (*InstanceCommandPayload, error) {
+	commandType := strings.TrimSpace(req.CommandType)
+	if !isSupportedCommandType(commandType) {
+		return nil, fmt.Errorf("invalid instance command type")
+	}
+	idempotencyKey := strings.TrimSpace(req.IdempotencyKey)
+	if idempotencyKey == "" {
+		idempotencyKey = fmt.Sprintf("%s-%d", commandType, time.Now().UTC().UnixNano())
+	}
+
+	existing, err := s.commandRepo.GetByInstanceIdempotencyKey(instanceID, idempotencyKey)
+	if err != nil {
+		return nil, err
+	}
+	if existing != nil {
+		return commandPayloadFromModel(*existing)
+	}
+
+	payloadJSON, err := marshalOptionalJSON(req.Payload)
+	if err != nil {
+		return nil, fmt.Errorf("failed to encode command payload: %w", err)
+	}
+	timeoutSeconds := req.TimeoutSeconds
+	if timeoutSeconds <= 0 {
+		timeoutSeconds = 300
+	}
+
+	now := time.Now().UTC()
+	command := &models.InstanceCommand{
+		InstanceID:     instanceID,
+		CommandType:    commandType,
+		PayloadJSON:    payloadJSON,
+		Status:         instanceCommandStatusPending,
+		IdempotencyKey: idempotencyKey,
+		IssuedBy:       issuedBy,
+		IssuedAt:       now,
+		TimeoutSeconds: timeoutSeconds,
+		CreatedAt:      now,
+		UpdatedAt:      now,
+	}
+	if err := s.commandRepo.Create(command); err != nil {
+		return nil, err
+	}
+
+	if err := s.applyDesiredStateSideEffects(instanceID, commandType, req.Payload); err != nil {
+		return nil, err
+	}
+	return commandPayloadFromModel(*command)
+}
+
+func (s *instanceCommandService) GetNextForAgent(session *AgentSession) (*AgentCommandEnvelope, error) {
+	command, err := s.commandRepo.GetNextPendingByInstance(session.Instance.ID)
+	if err != nil {
+		return nil, err
+	}
+	if command == nil {
+		return nil, nil
+	}
+
+	now := time.Now().UTC()
+	command.Status = instanceCommandStatusDispatched
+	command.AgentID = &session.Agent.AgentID
+	command.DispatchedAt = &now
+	if err := s.commandRepo.Update(command); err != nil {
+		return nil, err
+	}
+
+	payload := map[string]interface{}{}
+	if command.PayloadJSON != nil && strings.TrimSpace(*command.PayloadJSON) != "" {
+		if err := json.Unmarshal([]byte(*command.PayloadJSON), &payload); err != nil {
+			return nil, fmt.Errorf("failed to decode command payload: %w", err)
+		}
+	}
+
+	return &AgentCommandEnvelope{
+		ID:             command.ID,
+		CommandType:    command.CommandType,
+		Payload:        payload,
+		IssuedAt:       command.IssuedAt,
+		TimeoutSeconds: command.TimeoutSeconds,
+		IdempotencyKey: command.IdempotencyKey,
+	}, nil
+}
+
+func (s *instanceCommandService) MarkStarted(session *AgentSession, commandID int, startedAt *time.Time) error {
+	command, err := s.commandRepo.GetByID(commandID)
+	if err != nil {
+		return err
+	}
+	if command == nil || command.InstanceID != session.Instance.ID {
+		return fmt.Errorf("instance command not found")
+	}
+
+	if command.Status == instanceCommandStatusRunning || command.Status == instanceCommandStatusSucceeded || command.Status == instanceCommandStatusFailed {
+		return nil
+	}
+	now := time.Now().UTC()
+	if startedAt == nil || startedAt.IsZero() {
+		startedAt = &now
+	}
+	command.Status = instanceCommandStatusRunning
+	command.AgentID = &session.Agent.AgentID
+	command.StartedAt = startedAt
+	return s.commandRepo.Update(command)
+}
+
+func (s *instanceCommandService) MarkFinished(session *AgentSession, commandID int, req AgentCommandFinishRequest) error {
+	command, err := s.commandRepo.GetByID(commandID)
+	if err != nil {
+		return err
+	}
+	if command == nil || command.InstanceID != session.Instance.ID {
+		return fmt.Errorf("instance command not found")
+	}
+	if strings.TrimSpace(req.AgentID) != session.Agent.AgentID {
+		return fmt.Errorf("agent id does not match session")
+	}
+	if req.Status != instanceCommandStatusSucceeded && req.Status != instanceCommandStatusFailed {
+		return fmt.Errorf("invalid instance command finish status")
+	}
+
+	if command.Status == instanceCommandStatusSucceeded || command.Status == instanceCommandStatusFailed {
+		return nil
+	}
+
+	finishedAt := req.FinishedAt
+	if finishedAt == nil || finishedAt.IsZero() {
+		now := time.Now().UTC()
+		finishedAt = &now
+	}
+	command.Status = req.Status
+	command.AgentID = &session.Agent.AgentID
+	command.FinishedAt = finishedAt
+	if req.Result != nil {
+		resultJSON, err := marshalOptionalJSON(req.Result)
+		if err != nil {
+			return fmt.Errorf("failed to encode command result: %w", err)
+		}
+		command.ResultJSON = resultJSON
+	}
+	if strings.TrimSpace(req.ErrorMessage) != "" {
+		command.ErrorMessage = optionalString(strings.TrimSpace(req.ErrorMessage))
+	}
+	return s.commandRepo.Update(command)
+}
+
+func (s *instanceCommandService) ListByInstanceID(instanceID int, limit int) ([]InstanceCommandPayload, error) {
+	items, err := s.commandRepo.ListByInstanceID(instanceID, limit)
+	if err != nil {
+		return nil, err
+	}
+	result := make([]InstanceCommandPayload, 0, len(items))
+	for _, item := range items {
+		payload, err := commandPayloadFromModel(item)
+		if err != nil {
+			return nil, err
+		}
+		result = append(result, *payload)
+	}
+	return result, nil
+}
+
+func commandPayloadFromModel(item models.InstanceCommand) (*InstanceCommandPayload, error) {
+	payload := &InstanceCommandPayload{
+		ID:             item.ID,
+		CommandType:    item.CommandType,
+		Status:         item.Status,
+		IdempotencyKey: item.IdempotencyKey,
+		IssuedBy:       item.IssuedBy,
+		IssuedAt:       item.IssuedAt,
+		DispatchedAt:   item.DispatchedAt,
+		StartedAt:      item.StartedAt,
+		FinishedAt:     item.FinishedAt,
+		TimeoutSeconds: item.TimeoutSeconds,
+		ErrorMessage:   item.ErrorMessage,
+	}
+	if item.PayloadJSON != nil && strings.TrimSpace(*item.PayloadJSON) != "" {
+		if err := json.Unmarshal([]byte(*item.PayloadJSON), &payload.Payload); err != nil {
+			return nil, fmt.Errorf("failed to decode command payload: %w", err)
+		}
+	}
+	if item.ResultJSON != nil && strings.TrimSpace(*item.ResultJSON) != "" {
+		if err := json.Unmarshal([]byte(*item.ResultJSON), &payload.Result); err != nil {
+			return nil, fmt.Errorf("failed to decode command result: %w", err)
+		}
+	}
+	return payload, nil
+}
+
+func isSupportedCommandType(commandType string) bool {
+	switch strings.TrimSpace(commandType) {
+	case InstanceCommandTypeStartOpenClaw,
+		InstanceCommandTypeStopOpenClaw,
+		InstanceCommandTypeRestartOpenClaw,
+		InstanceCommandTypeCollectSystemInfo,
+		InstanceCommandTypeApplyConfigRevision,
+		InstanceCommandTypeReloadConfig,
+		InstanceCommandTypeHealthCheck:
+		return true
+	default:
+		return false
+	}
+}
+
+func (s *instanceCommandService) applyDesiredStateSideEffects(instanceID int, commandType string, payload map[string]interface{}) error {
+	state, err := s.desiredStateRepo.GetByInstanceID(instanceID)
+	if err != nil {
+		return err
+	}
+	if state == nil {
+		return nil
+	}
+
+	updated := false
+	switch commandType {
+	case InstanceCommandTypeStartOpenClaw, InstanceCommandTypeRestartOpenClaw:
+		if state.DesiredPowerState != "running" {
+			state.DesiredPowerState = "running"
+			updated = true
+		}
+	case InstanceCommandTypeStopOpenClaw:
+		if state.DesiredPowerState != "stopped" {
+			state.DesiredPowerState = "stopped"
+			updated = true
+		}
+	case InstanceCommandTypeApplyConfigRevision:
+		if revisionID, ok := payload["revision_id"].(float64); ok {
+			rev := int(revisionID)
+			state.DesiredConfigRevisionID = &rev
+			updated = true
+		}
+	}
+	if !updated {
+		return nil
+	}
+
+	state.UpdatedAt = time.Now().UTC()
+	if err := s.desiredStateRepo.Update(state); err != nil {
+		return err
+	}
+
+	runtime, err := s.runtimeRepo.GetByInstanceID(instanceID)
+	if err == nil && runtime != nil {
+		runtime.DesiredConfigRevisionID = state.DesiredConfigRevisionID
+		_ = s.runtimeRepo.Update(runtime)
+	}
+	return nil
+}