diff --git a/backend/cmd/server/main.go b/backend/cmd/server/main.go index e169049..b854ab4 100644 --- a/backend/cmd/server/main.go +++ b/backend/cmd/server/main.go @@ -54,6 +54,11 @@ func main() { riskRuleRepo := repository.NewRiskRuleRepository(database) riskHitRepo := repository.NewRiskHitRepository(database) openClawConfigRepo := repository.NewOpenClawConfigRepository(database) + instanceAgentRepo := repository.NewInstanceAgentRepository(database) + instanceRuntimeStatusRepo := repository.NewInstanceRuntimeStatusRepository(database) + instanceDesiredStateRepo := repository.NewInstanceDesiredStateRepository(database) + instanceCommandRepo := repository.NewInstanceCommandRepository(database) + instanceConfigRevisionRepo := repository.NewInstanceConfigRevisionRepository(database) if repaired, repairErr := services.RepairSeededAdminPassword(userRepo); repairErr != nil { log.Printf("Warning: failed to repair seeded admin password: %v", repairErr) @@ -80,12 +85,16 @@ func main() { clusterResourceService := services.NewClusterResourceService(instanceRepo) services.SetRuntimeImageSettingsProvider(systemImageSettingService) instanceService := services.NewInstanceService(instanceRepo, quotaRepo, llmModelRepo, openClawConfigService) + instanceAgentService := services.NewInstanceAgentService(instanceRepo, instanceAgentRepo, instanceDesiredStateRepo, instanceRuntimeStatusRepo, instanceCommandRepo) + instanceRuntimeStatusService := services.NewInstanceRuntimeStatusService(instanceRuntimeStatusRepo, instanceAgentRepo, instanceDesiredStateRepo) + instanceCommandService := services.NewInstanceCommandService(instanceCommandRepo, instanceRuntimeStatusRepo, instanceDesiredStateRepo) + instanceConfigRevisionService := services.NewInstanceConfigRevisionService(instanceConfigRevisionRepo) aiGatewayService := aigateway.NewService(llmModelRepo, modelInvocationService, auditEventService, costRecordService, riskDetectionService, riskHitService, chatSessionService, chatMessageService) // Initialize handlers authHandler := handlers.NewAuthHandler(authService) userHandler := handlers.NewUserHandler(userService, quotaService) - instanceHandler := handlers.NewInstanceHandler(instanceService) + instanceHandler := handlers.NewInstanceHandler(instanceService, instanceAgentService, instanceRuntimeStatusService, instanceCommandService, instanceConfigRevisionService, openClawConfigService) systemSettingsHandler := handlers.NewSystemSettingsHandler(systemImageSettingService) llmModelHandler := handlers.NewLLMModelHandler(llmModelService) aiGatewayHandler := handlers.NewAIGatewayHandler(aiGatewayService) @@ -94,13 +103,14 @@ func main() { clusterResourceHandler := handlers.NewClusterResourceHandler(clusterResourceService) egressProxyHandler := handlers.NewEgressProxyHandler() openClawConfigHandler := handlers.NewOpenClawConfigHandler(openClawConfigService) + agentHandler := handlers.NewAgentHandler(instanceAgentService, instanceCommandService, instanceRuntimeStatusService, instanceConfigRevisionService) // Initialize WebSocket hub and handler wsHub := services.GetHub() wsHandler := handlers.NewWebSocketHandler(wsHub) // Start sync service to keep instance status in sync with K8s - syncService := services.NewSyncService(instanceRepo) + syncService := services.NewSyncService(instanceRepo, instanceRuntimeStatusService) syncService.Start() defer syncService.Stop() @@ -164,6 +174,10 @@ func main() { instances.POST("/:id/stop", instanceHandler.StopInstance) instances.POST("/:id/restart", instanceHandler.RestartInstance) instances.GET("/:id/status", instanceHandler.GetInstanceStatus) + instances.GET("/:id/runtime", instanceHandler.GetRuntimeDetails) + instances.POST("/:id/runtime/:command", instanceHandler.CreateRuntimeCommand) + instances.GET("/:id/config/revisions", instanceHandler.ListConfigRevisions) + instances.POST("/:id/config/revisions/publish", instanceHandler.PublishConfigRevision) instances.POST("/:id/access", instanceHandler.GenerateAccessToken) instances.GET("/:id/access", instanceHandler.AccessInstance) instances.POST("/:id/sync", instanceHandler.ForceSync) @@ -259,6 +273,17 @@ func main() { gatewayLLM.POST("/chat/completions", aiGatewayHandler.ChatCompletions) } + agent := api.Group("/agent") + { + agent.POST("/register", agentHandler.Register) + agent.POST("/heartbeat", agentHandler.Heartbeat) + agent.GET("/commands/next", agentHandler.NextCommand) + agent.POST("/commands/:id/start", agentHandler.StartCommand) + agent.POST("/commands/:id/finish", agentHandler.FinishCommand) + agent.POST("/state/report", agentHandler.ReportState) + agent.GET("/config/revisions/:id", agentHandler.GetConfigRevision) + } + // Instance proxy routes (token-based auth, no session required) // These routes proxy requests to the actual instance pods api.Any("/instances/:id/proxy", instanceHandler.ProxyInstance) diff --git a/backend/internal/db/migrations/007_add_instance_agent_control_plane.sql b/backend/internal/db/migrations/007_add_instance_agent_control_plane.sql new file mode 100644 index 0000000..a18e7e2 --- /dev/null +++ b/backend/internal/db/migrations/007_add_instance_agent_control_plane.sql @@ -0,0 +1,123 @@ +SET @instance_agent_bootstrap_token_column_exists = ( + SELECT COUNT(*) + FROM information_schema.COLUMNS + WHERE TABLE_SCHEMA = DATABASE() + AND TABLE_NAME = 'instances' + AND COLUMN_NAME = 'agent_bootstrap_token' +); +SET @instance_agent_bootstrap_token_column_sql = IF( + @instance_agent_bootstrap_token_column_exists = 0, + 'ALTER TABLE instances ADD COLUMN agent_bootstrap_token VARCHAR(255) NULL AFTER access_token', + 'SELECT 1' +); +PREPARE instance_agent_bootstrap_token_column_stmt FROM @instance_agent_bootstrap_token_column_sql; +EXECUTE instance_agent_bootstrap_token_column_stmt; +DEALLOCATE PREPARE instance_agent_bootstrap_token_column_stmt; + +CREATE TABLE IF NOT EXISTS instance_agents ( + id INT AUTO_INCREMENT PRIMARY KEY, + instance_id INT NOT NULL, + agent_id VARCHAR(255) NOT NULL, + agent_version VARCHAR(50) NOT NULL, + protocol_version VARCHAR(50) NOT NULL, + status VARCHAR(30) NOT NULL DEFAULT 'online', + capabilities_json LONGTEXT NOT NULL, + host_info_json LONGTEXT NULL, + session_token VARCHAR(255) NULL, + session_expires_at TIMESTAMP NULL, + last_heartbeat_at TIMESTAMP NULL, + last_reported_at TIMESTAMP NULL, + last_seen_ip VARCHAR(45) NULL, + registered_at TIMESTAMP NULL, + created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, + FOREIGN KEY (instance_id) REFERENCES instances(id) ON DELETE CASCADE, + UNIQUE KEY uk_instance_agents_instance (instance_id), + UNIQUE KEY uk_instance_agents_session_token (session_token), + INDEX idx_instance_agents_agent_id (agent_id), + INDEX idx_instance_agents_status (status), + INDEX idx_instance_agents_last_heartbeat (last_heartbeat_at) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci; + +CREATE TABLE IF NOT EXISTS instance_runtime_status ( + id INT AUTO_INCREMENT PRIMARY KEY, + instance_id INT NOT NULL, + infra_status VARCHAR(30) NOT NULL DEFAULT 'creating', + agent_status VARCHAR(30) NOT NULL DEFAULT 'offline', + openclaw_status VARCHAR(30) NOT NULL DEFAULT 'unknown', + openclaw_pid INT NULL, + openclaw_version VARCHAR(100) NULL, + current_config_revision_id INT NULL, + desired_config_revision_id INT NULL, + summary_json LONGTEXT NULL, + system_info_json LONGTEXT NULL, + health_json LONGTEXT NULL, + last_reported_at TIMESTAMP NULL, + created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, + FOREIGN KEY (instance_id) REFERENCES instances(id) ON DELETE CASCADE, + UNIQUE KEY uk_instance_runtime_status_instance (instance_id), + INDEX idx_instance_runtime_status_agent_status (agent_status), + INDEX idx_instance_runtime_status_openclaw_status (openclaw_status) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci; + +CREATE TABLE IF NOT EXISTS instance_desired_state ( + id INT AUTO_INCREMENT PRIMARY KEY, + instance_id INT NOT NULL, + desired_power_state VARCHAR(30) NOT NULL DEFAULT 'running', + desired_config_revision_id INT NULL, + desired_runtime_action VARCHAR(50) NULL, + updated_by INT NULL, + updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, + created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + FOREIGN KEY (instance_id) REFERENCES instances(id) ON DELETE CASCADE, + FOREIGN KEY (updated_by) REFERENCES users(id) ON DELETE SET NULL, + UNIQUE KEY uk_instance_desired_state_instance (instance_id) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci; + +CREATE TABLE IF NOT EXISTS instance_commands ( + id INT AUTO_INCREMENT PRIMARY KEY, + instance_id INT NOT NULL, + agent_id VARCHAR(255) NULL, + command_type VARCHAR(50) NOT NULL, + payload_json LONGTEXT NULL, + status VARCHAR(30) NOT NULL DEFAULT 'pending', + idempotency_key VARCHAR(255) NOT NULL, + issued_by INT NULL, + issued_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + dispatched_at TIMESTAMP NULL, + started_at TIMESTAMP NULL, + finished_at TIMESTAMP NULL, + timeout_seconds INT NOT NULL DEFAULT 300, + result_json LONGTEXT NULL, + error_message TEXT NULL, + created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, + FOREIGN KEY (instance_id) REFERENCES instances(id) ON DELETE CASCADE, + FOREIGN KEY (issued_by) REFERENCES users(id) ON DELETE SET NULL, + UNIQUE KEY uk_instance_commands_idempotency (instance_id, idempotency_key), + INDEX idx_instance_commands_instance_status (instance_id, status), + INDEX idx_instance_commands_agent_status (agent_id, status), + INDEX idx_instance_commands_issued_at (issued_at) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci; + +CREATE TABLE IF NOT EXISTS instance_config_revisions ( + id INT AUTO_INCREMENT PRIMARY KEY, + instance_id INT NOT NULL, + source_snapshot_id INT NULL, + source_bundle_id INT NULL, + revision_no INT NOT NULL, + content_json LONGTEXT NOT NULL, + checksum VARCHAR(255) NOT NULL, + status VARCHAR(30) NOT NULL DEFAULT 'published', + published_by INT NULL, + published_at TIMESTAMP NULL, + activated_at TIMESTAMP NULL, + created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, + FOREIGN KEY (instance_id) REFERENCES instances(id) ON DELETE CASCADE, + FOREIGN KEY (published_by) REFERENCES users(id) ON DELETE SET NULL, + UNIQUE KEY uk_instance_config_revision_unique (instance_id, revision_no), + INDEX idx_instance_config_revision_instance (instance_id, revision_no), + INDEX idx_instance_config_revision_status (status) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci; diff --git a/backend/internal/handlers/agent_handler.go b/backend/internal/handlers/agent_handler.go new file mode 100644 index 0000000..8347ccf --- /dev/null +++ b/backend/internal/handlers/agent_handler.go @@ -0,0 +1,205 @@ +package handlers + +import ( + "net/http" + "strconv" + "strings" + "time" + + "clawreef/internal/services" + "clawreef/internal/utils" + + "github.com/gin-gonic/gin" +) + +type AgentHandler struct { + agentService services.InstanceAgentService + commandService services.InstanceCommandService + runtimeStatusService services.InstanceRuntimeStatusService + configRevisionService services.InstanceConfigRevisionService +} + +func NewAgentHandler(agentService services.InstanceAgentService, commandService services.InstanceCommandService, runtimeStatusService services.InstanceRuntimeStatusService, configRevisionService services.InstanceConfigRevisionService) *AgentHandler { + return &AgentHandler{ + agentService: agentService, + commandService: commandService, + runtimeStatusService: runtimeStatusService, + configRevisionService: configRevisionService, + } +} + +func (h *AgentHandler) Register(c *gin.Context) { + bootstrapToken := extractBearerToken(c.GetHeader("Authorization")) + if bootstrapToken == "" { + utils.Error(c, http.StatusUnauthorized, "Agent bootstrap token is required") + return + } + + var req services.AgentRegisterRequest + if err := c.ShouldBindJSON(&req); err != nil { + utils.ValidationError(c, err) + return + } + + resp, err := h.agentService.Register(bootstrapToken, req, c.ClientIP()) + if err != nil { + utils.HandleError(c, err) + return + } + utils.Success(c, http.StatusOK, "Agent registered successfully", resp) +} + +func (h *AgentHandler) Heartbeat(c *gin.Context) { + session, ok := h.authenticateAgentSession(c) + if !ok { + return + } + + var req services.AgentHeartbeatRequest + if err := c.ShouldBindJSON(&req); err != nil { + utils.ValidationError(c, err) + return + } + if req.Timestamp.IsZero() { + req.Timestamp = time.Now().UTC() + } + + resp, err := h.agentService.Heartbeat(session, req, c.ClientIP()) + if err != nil { + utils.HandleError(c, err) + return + } + utils.Success(c, http.StatusOK, "Agent heartbeat accepted", resp) +} + +func (h *AgentHandler) NextCommand(c *gin.Context) { + session, ok := h.authenticateAgentSession(c) + if !ok { + return + } + + command, err := h.commandService.GetNextForAgent(session) + if err != nil { + utils.HandleError(c, err) + return + } + utils.Success(c, http.StatusOK, "Agent next command retrieved successfully", gin.H{"command": command}) +} + +func (h *AgentHandler) StartCommand(c *gin.Context) { + session, ok := h.authenticateAgentSession(c) + if !ok { + return + } + commandID, err := strconv.Atoi(c.Param("id")) + if err != nil { + utils.Error(c, http.StatusBadRequest, "Invalid command ID") + return + } + var req struct { + AgentID string `json:"agent_id" binding:"required"` + StartedAt *time.Time `json:"started_at,omitempty"` + } + if err := c.ShouldBindJSON(&req); err != nil { + utils.ValidationError(c, err) + return + } + if req.AgentID != session.Agent.AgentID { + utils.Error(c, http.StatusForbidden, "Agent ID does not match session") + return + } + if err := h.commandService.MarkStarted(session, commandID, req.StartedAt); err != nil { + utils.HandleError(c, err) + return + } + utils.Success(c, http.StatusOK, "Agent command marked as started", nil) +} + +func (h *AgentHandler) FinishCommand(c *gin.Context) { + session, ok := h.authenticateAgentSession(c) + if !ok { + return + } + commandID, err := strconv.Atoi(c.Param("id")) + if err != nil { + utils.Error(c, http.StatusBadRequest, "Invalid command ID") + return + } + + var req services.AgentCommandFinishRequest + if err := c.ShouldBindJSON(&req); err != nil { + utils.ValidationError(c, err) + return + } + if err := h.commandService.MarkFinished(session, commandID, req); err != nil { + utils.HandleError(c, err) + return + } + utils.Success(c, http.StatusOK, "Agent command result accepted", nil) +} + +func (h *AgentHandler) ReportState(c *gin.Context) { + session, ok := h.authenticateAgentSession(c) + if !ok { + return + } + + var req services.AgentStateReportRequest + if err := c.ShouldBindJSON(&req); err != nil { + utils.ValidationError(c, err) + return + } + if err := h.runtimeStatusService.Report(session, req, c.ClientIP()); err != nil { + utils.HandleError(c, err) + return + } + utils.Success(c, http.StatusOK, "Agent state reported successfully", nil) +} + +func (h *AgentHandler) GetConfigRevision(c *gin.Context) { + session, ok := h.authenticateAgentSession(c) + if !ok { + return + } + revisionID, err := strconv.Atoi(c.Param("id")) + if err != nil { + utils.Error(c, http.StatusBadRequest, "Invalid config revision ID") + return + } + revision, err := h.configRevisionService.GetByID(revisionID) + if err != nil { + utils.HandleError(c, err) + return + } + if revision.InstanceID != session.Instance.ID { + utils.Error(c, http.StatusForbidden, "Access denied") + return + } + utils.Success(c, http.StatusOK, "Config revision retrieved successfully", gin.H{"revision": revision}) +} + +func (h *AgentHandler) authenticateAgentSession(c *gin.Context) (*services.AgentSession, bool) { + sessionToken := extractBearerToken(c.GetHeader("Authorization")) + if sessionToken == "" { + utils.Error(c, http.StatusUnauthorized, "Agent session token is required") + return nil, false + } + session, err := h.agentService.AuthenticateSession(sessionToken) + if err != nil { + utils.HandleError(c, err) + return nil, false + } + return session, true +} + +func extractBearerToken(raw string) string { + raw = strings.TrimSpace(raw) + if raw == "" { + return "" + } + parts := strings.SplitN(raw, " ", 2) + if len(parts) != 2 || !strings.EqualFold(parts[0], "bearer") { + return "" + } + return strings.TrimSpace(parts[1]) +} diff --git a/backend/internal/handlers/instance_handler.go b/backend/internal/handlers/instance_handler.go index 22b9330..a0d95b5 100644 --- a/backend/internal/handlers/instance_handler.go +++ b/backend/internal/handlers/instance_handler.go @@ -17,23 +17,47 @@ import ( // InstanceHandler handles instance management requests type InstanceHandler struct { - instanceService services.InstanceService - accessService *services.InstanceAccessService - proxyService *services.InstanceProxyService - openClawTransferService services.OpenClawTransferService + instanceService services.InstanceService + instanceAgentService services.InstanceAgentService + runtimeStatusService services.InstanceRuntimeStatusService + instanceCommandService services.InstanceCommandService + instanceConfigRevisionService services.InstanceConfigRevisionService + accessService *services.InstanceAccessService + proxyService *services.InstanceProxyService + openClawTransferService services.OpenClawTransferService + openClawConfigService services.OpenClawConfigService } // NewInstanceHandler creates a new instance handler -func NewInstanceHandler(instanceService services.InstanceService) *InstanceHandler { +func NewInstanceHandler(instanceService services.InstanceService, instanceAgentService services.InstanceAgentService, runtimeStatusService services.InstanceRuntimeStatusService, instanceCommandService services.InstanceCommandService, instanceConfigRevisionService services.InstanceConfigRevisionService, openClawConfigService services.OpenClawConfigService) *InstanceHandler { accessService := services.NewInstanceAccessService() return &InstanceHandler{ - instanceService: instanceService, - accessService: accessService, - proxyService: services.NewInstanceProxyService(accessService), - openClawTransferService: services.NewOpenClawTransferService(), + instanceService: instanceService, + instanceAgentService: instanceAgentService, + runtimeStatusService: runtimeStatusService, + instanceCommandService: instanceCommandService, + instanceConfigRevisionService: instanceConfigRevisionService, + accessService: accessService, + proxyService: services.NewInstanceProxyService(accessService), + openClawTransferService: services.NewOpenClawTransferService(), + openClawConfigService: openClawConfigService, } } +type InstanceRuntimeDetailsResponse struct { + Runtime *services.InstanceRuntimeStatusPayload `json:"runtime,omitempty"` + Agent *services.InstanceAgentPayload `json:"agent,omitempty"` + Commands []services.InstanceCommandPayload `json:"commands,omitempty"` +} + +type CreateRuntimeCommandRequest struct { + IdempotencyKey string `json:"idempotency_key,omitempty"` +} + +type PublishConfigRevisionRequest struct { + SnapshotID int `json:"snapshot_id" binding:"required,min=1"` +} + // CreateInstanceRequest represents a create instance request type CreateInstanceRequest struct { Name string `json:"name" binding:"required,min=3,max=50"` @@ -68,6 +92,7 @@ type ListInstancesRequest struct { // ListInstances lists instances for the current user func (h *InstanceHandler) ListInstances(c *gin.Context) { userID, _ := c.Get("userID") + userRole, _ := c.Get("userRole") var req ListInstancesRequest if err := c.ShouldBindQuery(&req); err != nil { @@ -78,7 +103,7 @@ func (h *InstanceHandler) ListInstances(c *gin.Context) { // Calculate offset offset := (req.Page - 1) * req.Limit - instances, total, err := h.instanceService.GetByUserID(userID.(int), offset, req.Limit) + instances, total, err := h.instanceService.GetVisibleInstances(userID.(int), fmt.Sprintf("%v", userRole), offset, req.Limit) if err != nil { utils.HandleError(c, err) return @@ -158,7 +183,14 @@ func (h *InstanceHandler) GetInstance(c *gin.Context) { return } - utils.Success(c, http.StatusOK, "Instance retrieved successfully", instance) + runtime, _ := h.runtimeStatusService.GetByInstanceID(instance.ID) + agent, _ := h.instanceAgentService.GetPayloadByInstanceID(instance.ID) + + utils.Success(c, http.StatusOK, "Instance retrieved successfully", gin.H{ + "instance": instance, + "runtime": runtime, + "agent": agent, + }) } // UpdateInstance updates an instance @@ -392,7 +424,193 @@ func (h *InstanceHandler) GetInstanceStatus(c *gin.Context) { return } - utils.Success(c, http.StatusOK, "Instance status retrieved successfully", status) + runtime, _ := h.runtimeStatusService.GetByInstanceID(id) + agent, _ := h.instanceAgentService.GetPayloadByInstanceID(id) + + utils.Success(c, http.StatusOK, "Instance status retrieved successfully", gin.H{ + "instance_status": status, + "runtime": runtime, + "agent": agent, + }) +} + +func (h *InstanceHandler) GetRuntimeDetails(c *gin.Context) { + id, _, ok := h.resolveOwnedInstance(c) + if !ok { + return + } + + runtime, err := h.runtimeStatusService.GetByInstanceID(id) + if err != nil { + utils.HandleError(c, err) + return + } + agent, err := h.instanceAgentService.GetPayloadByInstanceID(id) + if err != nil { + utils.HandleError(c, err) + return + } + commands, err := h.instanceCommandService.ListByInstanceID(id, 20) + if err != nil { + utils.HandleError(c, err) + return + } + + utils.Success(c, http.StatusOK, "Instance runtime details retrieved successfully", InstanceRuntimeDetailsResponse{ + Runtime: runtime, + Agent: agent, + Commands: commands, + }) +} + +func (h *InstanceHandler) CreateRuntimeCommand(c *gin.Context) { + id, _, ok := h.resolveOwnedInstance(c) + if !ok { + return + } + + commandKey := strings.TrimSpace(c.Param("command")) + commandType := "" + switch commandKey { + case "start": + commandType = services.InstanceCommandTypeStartOpenClaw + case "stop": + commandType = services.InstanceCommandTypeStopOpenClaw + case "restart": + commandType = services.InstanceCommandTypeRestartOpenClaw + case "collect-system-info": + commandType = services.InstanceCommandTypeCollectSystemInfo + case "health-check": + commandType = services.InstanceCommandTypeHealthCheck + default: + utils.Error(c, http.StatusBadRequest, "Invalid runtime command") + return + } + + var req CreateRuntimeCommandRequest + if err := c.ShouldBindJSON(&req); err != nil && err != io.EOF { + utils.ValidationError(c, err) + return + } + + userID, _ := c.Get("userID") + issuedBy := userID.(int) + command, err := h.instanceCommandService.Create(id, &issuedBy, services.CreateInstanceCommandRequest{ + CommandType: commandType, + IdempotencyKey: req.IdempotencyKey, + }) + if err != nil { + utils.HandleError(c, err) + return + } + + utils.Success(c, http.StatusCreated, "Instance runtime command created successfully", command) +} + +func (h *InstanceHandler) ListConfigRevisions(c *gin.Context) { + id, _, ok := h.resolveOwnedInstance(c) + if !ok { + return + } + + limit, _ := strconv.Atoi(c.DefaultQuery("limit", "20")) + items, err := h.instanceConfigRevisionService.ListByInstanceID(id, limit) + if err != nil { + utils.HandleError(c, err) + return + } + + utils.Success(c, http.StatusOK, "Instance config revisions retrieved successfully", items) +} + +func (h *InstanceHandler) PublishConfigRevision(c *gin.Context) { + id, instance, ok := h.resolveOwnedInstance(c) + if !ok { + return + } + if !strings.EqualFold(instance.Type, "openclaw") { + utils.Error(c, http.StatusBadRequest, "Only openclaw instances support config revisions") + return + } + + var req PublishConfigRevisionRequest + if err := c.ShouldBindJSON(&req); err != nil { + utils.ValidationError(c, err) + return + } + + userID, _ := c.Get("userID") + snapshot, err := h.openClawConfigService.GetSnapshot(userID.(int), req.SnapshotID) + if err != nil { + utils.HandleError(c, err) + return + } + if snapshot.InstanceID != nil && *snapshot.InstanceID != id { + utils.Error(c, http.StatusBadRequest, "Snapshot does not belong to this instance") + return + } + + modelSnapshot := &models.OpenClawInjectionSnapshot{ + ID: snapshot.ID, + InstanceID: snapshot.InstanceID, + UserID: snapshot.UserID, + BundleID: snapshot.BundleID, + Mode: snapshot.Mode, + RenderedManifestJSON: string(snapshot.Manifest), + } + + issuedBy := userID.(int) + revision, err := h.instanceConfigRevisionService.CreateFromSnapshot(id, modelSnapshot, &issuedBy) + if err != nil { + utils.HandleError(c, err) + return + } + + command, err := h.instanceCommandService.Create(id, &issuedBy, services.CreateInstanceCommandRequest{ + CommandType: services.InstanceCommandTypeApplyConfigRevision, + IdempotencyKey: fmt.Sprintf("apply-config-revision-%d", revision.ID), + Payload: map[string]interface{}{ + "revision_id": revision.ID, + "snapshot_id": snapshot.ID, + }, + TimeoutSeconds: 300, + }) + if err != nil { + utils.HandleError(c, err) + return + } + + utils.Success(c, http.StatusCreated, "Instance config revision published successfully", gin.H{ + "revision": revision, + "command": command, + }) +} + +func (h *InstanceHandler) resolveOwnedInstance(c *gin.Context) (int, *models.Instance, bool) { + id, err := strconv.Atoi(c.Param("id")) + if err != nil { + utils.Error(c, http.StatusBadRequest, "Invalid instance ID") + return 0, nil, false + } + + instance, err := h.instanceService.GetByID(id) + if err != nil { + utils.HandleError(c, err) + return 0, nil, false + } + if instance == nil { + utils.Error(c, http.StatusNotFound, "Instance not found") + return 0, nil, false + } + + userID, _ := c.Get("userID") + userRole, _ := c.Get("userRole") + if userRole != "admin" && instance.UserID != userID.(int) { + utils.Error(c, http.StatusForbidden, "Access denied") + return 0, nil, false + } + + return id, instance, true } // GenerateAccessToken generates an access token for an instance diff --git a/backend/internal/models/instance.go b/backend/internal/models/instance.go index 5c9a84f..fdcce11 100644 --- a/backend/internal/models/instance.go +++ b/backend/internal/models/instance.go @@ -29,6 +29,7 @@ type Instance struct { PodIP *string `db:"pod_ip" json:"pod_ip,omitempty"` AccessURL *string `db:"access_url" json:"access_url,omitempty"` AccessToken *string `db:"access_token" json:"-"` + AgentBootstrapToken *string `db:"agent_bootstrap_token" json:"-"` OpenClawConfigSnapshotID *int `db:"openclaw_config_snapshot_id" json:"openclaw_config_snapshot_id,omitempty"` CreatedAt time.Time `db:"created_at" json:"created_at"` UpdatedAt time.Time `db:"updated_at" json:"updated_at"` diff --git a/backend/internal/models/instance_agent.go b/backend/internal/models/instance_agent.go new file mode 100644 index 0000000..556f980 --- /dev/null +++ b/backend/internal/models/instance_agent.go @@ -0,0 +1,26 @@ +package models + +import "time" + +type InstanceAgent struct { + ID int `db:"id,primarykey,autoincrement" json:"id"` + InstanceID int `db:"instance_id" json:"instance_id"` + AgentID string `db:"agent_id" json:"agent_id"` + AgentVersion string `db:"agent_version" json:"agent_version"` + ProtocolVersion string `db:"protocol_version" json:"protocol_version"` + Status string `db:"status" json:"status"` + CapabilitiesJSON string `db:"capabilities_json" json:"-"` + HostInfoJSON *string `db:"host_info_json" json:"-"` + SessionToken *string `db:"session_token" json:"-"` + SessionExpiresAt *time.Time `db:"session_expires_at" json:"session_expires_at,omitempty"` + LastHeartbeatAt *time.Time `db:"last_heartbeat_at" json:"last_heartbeat_at,omitempty"` + LastReportedAt *time.Time `db:"last_reported_at" json:"last_reported_at,omitempty"` + LastSeenIP *string `db:"last_seen_ip" json:"last_seen_ip,omitempty"` + RegisteredAt *time.Time `db:"registered_at" json:"registered_at,omitempty"` + CreatedAt time.Time `db:"created_at" json:"created_at"` + UpdatedAt time.Time `db:"updated_at" json:"updated_at"` +} + +func (a InstanceAgent) TableName() string { + return "instance_agents" +} diff --git a/backend/internal/models/instance_command.go b/backend/internal/models/instance_command.go new file mode 100644 index 0000000..c310cb2 --- /dev/null +++ b/backend/internal/models/instance_command.go @@ -0,0 +1,27 @@ +package models + +import "time" + +type InstanceCommand struct { + ID int `db:"id,primarykey,autoincrement" json:"id"` + InstanceID int `db:"instance_id" json:"instance_id"` + AgentID *string `db:"agent_id" json:"agent_id,omitempty"` + CommandType string `db:"command_type" json:"command_type"` + PayloadJSON *string `db:"payload_json" json:"-"` + Status string `db:"status" json:"status"` + IdempotencyKey string `db:"idempotency_key" json:"idempotency_key"` + IssuedBy *int `db:"issued_by" json:"issued_by,omitempty"` + IssuedAt time.Time `db:"issued_at" json:"issued_at"` + DispatchedAt *time.Time `db:"dispatched_at" json:"dispatched_at,omitempty"` + StartedAt *time.Time `db:"started_at" json:"started_at,omitempty"` + FinishedAt *time.Time `db:"finished_at" json:"finished_at,omitempty"` + TimeoutSeconds int `db:"timeout_seconds" json:"timeout_seconds"` + ResultJSON *string `db:"result_json" json:"-"` + ErrorMessage *string `db:"error_message" json:"error_message,omitempty"` + CreatedAt time.Time `db:"created_at" json:"created_at"` + UpdatedAt time.Time `db:"updated_at" json:"updated_at"` +} + +func (c InstanceCommand) TableName() string { + return "instance_commands" +} diff --git a/backend/internal/models/instance_config_revision.go b/backend/internal/models/instance_config_revision.go new file mode 100644 index 0000000..01678f7 --- /dev/null +++ b/backend/internal/models/instance_config_revision.go @@ -0,0 +1,23 @@ +package models + +import "time" + +type InstanceConfigRevision struct { + ID int `db:"id,primarykey,autoincrement" json:"id"` + InstanceID int `db:"instance_id" json:"instance_id"` + SourceSnapshotID *int `db:"source_snapshot_id" json:"source_snapshot_id,omitempty"` + SourceBundleID *int `db:"source_bundle_id" json:"source_bundle_id,omitempty"` + RevisionNo int `db:"revision_no" json:"revision_no"` + ContentJSON string `db:"content_json" json:"-"` + Checksum string `db:"checksum" json:"checksum"` + Status string `db:"status" json:"status"` + PublishedBy *int `db:"published_by" json:"published_by,omitempty"` + PublishedAt *time.Time `db:"published_at" json:"published_at,omitempty"` + ActivatedAt *time.Time `db:"activated_at" json:"activated_at,omitempty"` + CreatedAt time.Time `db:"created_at" json:"created_at"` + UpdatedAt time.Time `db:"updated_at" json:"updated_at"` +} + +func (r InstanceConfigRevision) TableName() string { + return "instance_config_revisions" +} diff --git a/backend/internal/models/instance_desired_state.go b/backend/internal/models/instance_desired_state.go new file mode 100644 index 0000000..2df54e1 --- /dev/null +++ b/backend/internal/models/instance_desired_state.go @@ -0,0 +1,18 @@ +package models + +import "time" + +type InstanceDesiredState struct { + ID int `db:"id,primarykey,autoincrement" json:"id"` + InstanceID int `db:"instance_id" json:"instance_id"` + DesiredPowerState string `db:"desired_power_state" json:"desired_power_state"` + DesiredConfigRevisionID *int `db:"desired_config_revision_id" json:"desired_config_revision_id,omitempty"` + DesiredRuntimeAction *string `db:"desired_runtime_action" json:"desired_runtime_action,omitempty"` + UpdatedBy *int `db:"updated_by" json:"updated_by,omitempty"` + UpdatedAt time.Time `db:"updated_at" json:"updated_at"` + CreatedAt time.Time `db:"created_at" json:"created_at"` +} + +func (s InstanceDesiredState) TableName() string { + return "instance_desired_state" +} diff --git a/backend/internal/models/instance_runtime_status.go b/backend/internal/models/instance_runtime_status.go new file mode 100644 index 0000000..b38f2c4 --- /dev/null +++ b/backend/internal/models/instance_runtime_status.go @@ -0,0 +1,25 @@ +package models + +import "time" + +type InstanceRuntimeStatus struct { + ID int `db:"id,primarykey,autoincrement" json:"id"` + InstanceID int `db:"instance_id" json:"instance_id"` + InfraStatus string `db:"infra_status" json:"infra_status"` + AgentStatus string `db:"agent_status" json:"agent_status"` + OpenClawStatus string `db:"openclaw_status" json:"openclaw_status"` + OpenClawPID *int `db:"openclaw_pid" json:"openclaw_pid,omitempty"` + OpenClawVersion *string `db:"openclaw_version" json:"openclaw_version,omitempty"` + CurrentConfigRevisionID *int `db:"current_config_revision_id" json:"current_config_revision_id,omitempty"` + DesiredConfigRevisionID *int `db:"desired_config_revision_id" json:"desired_config_revision_id,omitempty"` + SummaryJSON *string `db:"summary_json" json:"-"` + SystemInfoJSON *string `db:"system_info_json" json:"-"` + HealthJSON *string `db:"health_json" json:"-"` + LastReportedAt *time.Time `db:"last_reported_at" json:"last_reported_at,omitempty"` + CreatedAt time.Time `db:"created_at" json:"created_at"` + UpdatedAt time.Time `db:"updated_at" json:"updated_at"` +} + +func (s InstanceRuntimeStatus) TableName() string { + return "instance_runtime_status" +} diff --git a/backend/internal/repository/instance_agent_repository.go b/backend/internal/repository/instance_agent_repository.go new file mode 100644 index 0000000..b1fffbe --- /dev/null +++ b/backend/internal/repository/instance_agent_repository.go @@ -0,0 +1,69 @@ +package repository + +import ( + "fmt" + "time" + + "clawreef/internal/models" + + "github.com/upper/db/v4" +) + +type InstanceAgentRepository interface { + GetByInstanceID(instanceID int) (*models.InstanceAgent, error) + GetBySessionToken(sessionToken string) (*models.InstanceAgent, error) + Create(agent *models.InstanceAgent) error + Update(agent *models.InstanceAgent) error +} + +type instanceAgentRepository struct { + sess db.Session +} + +func NewInstanceAgentRepository(sess db.Session) InstanceAgentRepository { + return &instanceAgentRepository{sess: sess} +} + +func (r *instanceAgentRepository) GetByInstanceID(instanceID int) (*models.InstanceAgent, error) { + var item models.InstanceAgent + if err := r.sess.Collection("instance_agents").Find(db.Cond{"instance_id": instanceID}).One(&item); err != nil { + if err == db.ErrNoMoreRows { + return nil, nil + } + return nil, fmt.Errorf("failed to get instance agent: %w", err) + } + return &item, nil +} + +func (r *instanceAgentRepository) GetBySessionToken(sessionToken string) (*models.InstanceAgent, error) { + var item models.InstanceAgent + if err := r.sess.Collection("instance_agents").Find(db.Cond{"session_token": sessionToken}).One(&item); err != nil { + if err == db.ErrNoMoreRows { + return nil, nil + } + return nil, fmt.Errorf("failed to get instance agent by session token: %w", err) + } + return &item, nil +} + +func (r *instanceAgentRepository) Create(agent *models.InstanceAgent) error { + ensureTimestamps(&agent.CreatedAt, &agent.UpdatedAt) + res, err := r.sess.Collection("instance_agents").Insert(agent) + if err != nil { + return fmt.Errorf("failed to create instance agent: %w", err) + } + if id, ok := res.ID().(int64); ok { + agent.ID = int(id) + } + return nil +} + +func (r *instanceAgentRepository) Update(agent *models.InstanceAgent) error { + if agent.UpdatedAt.IsZero() { + agent.UpdatedAt = time.Now().UTC() + } + if err := r.sess.Collection("instance_agents").Find(db.Cond{"id": agent.ID}).Update(agent); err != nil { + return fmt.Errorf("failed to update instance agent: %w", err) + } + return nil +} diff --git a/backend/internal/repository/instance_command_repository.go b/backend/internal/repository/instance_command_repository.go new file mode 100644 index 0000000..29032f7 --- /dev/null +++ b/backend/internal/repository/instance_command_repository.go @@ -0,0 +1,97 @@ +package repository + +import ( + "fmt" + "time" + + "clawreef/internal/models" + + "github.com/upper/db/v4" +) + +type InstanceCommandRepository interface { + Create(command *models.InstanceCommand) error + Update(command *models.InstanceCommand) error + GetByID(id int) (*models.InstanceCommand, error) + GetByInstanceIdempotencyKey(instanceID int, idempotencyKey string) (*models.InstanceCommand, error) + GetNextPendingByInstance(instanceID int) (*models.InstanceCommand, error) + ListByInstanceID(instanceID int, limit int) ([]models.InstanceCommand, error) +} + +type instanceCommandRepository struct { + sess db.Session +} + +func NewInstanceCommandRepository(sess db.Session) InstanceCommandRepository { + return &instanceCommandRepository{sess: sess} +} + +func (r *instanceCommandRepository) Create(command *models.InstanceCommand) error { + ensureTimestamps(&command.CreatedAt, &command.UpdatedAt) + if command.IssuedAt.IsZero() { + command.IssuedAt = time.Now().UTC() + } + res, err := r.sess.Collection("instance_commands").Insert(command) + if err != nil { + return fmt.Errorf("failed to create instance command: %w", err) + } + if id, ok := res.ID().(int64); ok { + command.ID = int(id) + } + return nil +} + +func (r *instanceCommandRepository) Update(command *models.InstanceCommand) error { + if command.UpdatedAt.IsZero() { + command.UpdatedAt = time.Now().UTC() + } + if err := r.sess.Collection("instance_commands").Find(db.Cond{"id": command.ID}).Update(command); err != nil { + return fmt.Errorf("failed to update instance command: %w", err) + } + return nil +} + +func (r *instanceCommandRepository) GetByID(id int) (*models.InstanceCommand, error) { + var item models.InstanceCommand + if err := r.sess.Collection("instance_commands").Find(db.Cond{"id": id}).One(&item); err != nil { + if err == db.ErrNoMoreRows { + return nil, nil + } + return nil, fmt.Errorf("failed to get instance command: %w", err) + } + return &item, nil +} + +func (r *instanceCommandRepository) GetByInstanceIdempotencyKey(instanceID int, idempotencyKey string) (*models.InstanceCommand, error) { + var item models.InstanceCommand + if err := r.sess.Collection("instance_commands").Find(db.Cond{"instance_id": instanceID, "idempotency_key": idempotencyKey}).One(&item); err != nil { + if err == db.ErrNoMoreRows { + return nil, nil + } + return nil, fmt.Errorf("failed to get instance command by idempotency key: %w", err) + } + return &item, nil +} + +func (r *instanceCommandRepository) GetNextPendingByInstance(instanceID int) (*models.InstanceCommand, error) { + var item models.InstanceCommand + if err := r.sess.Collection("instance_commands").Find(db.Cond{"instance_id": instanceID, "status": "pending"}).OrderBy("issued_at", "id").One(&item); err != nil { + if err == db.ErrNoMoreRows { + return nil, nil + } + return nil, fmt.Errorf("failed to get next pending instance command: %w", err) + } + return &item, nil +} + +func (r *instanceCommandRepository) ListByInstanceID(instanceID int, limit int) ([]models.InstanceCommand, error) { + if limit <= 0 { + limit = 20 + } + + var items []models.InstanceCommand + if err := r.sess.Collection("instance_commands").Find(db.Cond{"instance_id": instanceID}).OrderBy("-issued_at", "-id").Limit(limit).All(&items); err != nil { + return nil, fmt.Errorf("failed to list instance commands: %w", err) + } + return items, nil +} diff --git a/backend/internal/repository/instance_config_revision_repository.go b/backend/internal/repository/instance_config_revision_repository.go new file mode 100644 index 0000000..0d2b60f --- /dev/null +++ b/backend/internal/repository/instance_config_revision_repository.go @@ -0,0 +1,82 @@ +package repository + +import ( + "fmt" + "time" + + "clawreef/internal/models" + + "github.com/upper/db/v4" +) + +type InstanceConfigRevisionRepository interface { + Create(revision *models.InstanceConfigRevision) error + Update(revision *models.InstanceConfigRevision) error + GetByID(id int) (*models.InstanceConfigRevision, error) + GetLatestByInstanceID(instanceID int) (*models.InstanceConfigRevision, error) + ListByInstanceID(instanceID int, limit int) ([]models.InstanceConfigRevision, error) +} + +type instanceConfigRevisionRepository struct { + sess db.Session +} + +func NewInstanceConfigRevisionRepository(sess db.Session) InstanceConfigRevisionRepository { + return &instanceConfigRevisionRepository{sess: sess} +} + +func (r *instanceConfigRevisionRepository) Create(revision *models.InstanceConfigRevision) error { + ensureTimestamps(&revision.CreatedAt, &revision.UpdatedAt) + res, err := r.sess.Collection("instance_config_revisions").Insert(revision) + if err != nil { + return fmt.Errorf("failed to create instance config revision: %w", err) + } + if id, ok := res.ID().(int64); ok { + revision.ID = int(id) + } + return nil +} + +func (r *instanceConfigRevisionRepository) Update(revision *models.InstanceConfigRevision) error { + if revision.UpdatedAt.IsZero() { + revision.UpdatedAt = time.Now().UTC() + } + if err := r.sess.Collection("instance_config_revisions").Find(db.Cond{"id": revision.ID}).Update(revision); err != nil { + return fmt.Errorf("failed to update instance config revision: %w", err) + } + return nil +} + +func (r *instanceConfigRevisionRepository) GetByID(id int) (*models.InstanceConfigRevision, error) { + var item models.InstanceConfigRevision + if err := r.sess.Collection("instance_config_revisions").Find(db.Cond{"id": id}).One(&item); err != nil { + if err == db.ErrNoMoreRows { + return nil, nil + } + return nil, fmt.Errorf("failed to get instance config revision: %w", err) + } + return &item, nil +} + +func (r *instanceConfigRevisionRepository) GetLatestByInstanceID(instanceID int) (*models.InstanceConfigRevision, error) { + var item models.InstanceConfigRevision + if err := r.sess.Collection("instance_config_revisions").Find(db.Cond{"instance_id": instanceID}).OrderBy("-revision_no", "-id").One(&item); err != nil { + if err == db.ErrNoMoreRows { + return nil, nil + } + return nil, fmt.Errorf("failed to get latest instance config revision: %w", err) + } + return &item, nil +} + +func (r *instanceConfigRevisionRepository) ListByInstanceID(instanceID int, limit int) ([]models.InstanceConfigRevision, error) { + if limit <= 0 { + limit = 20 + } + + var items []models.InstanceConfigRevision + if err := r.sess.Collection("instance_config_revisions").Find(db.Cond{"instance_id": instanceID}).OrderBy("-revision_no", "-id").Limit(limit).All(&items); err != nil { + return nil, fmt.Errorf("failed to list instance config revisions: %w", err) + } + return items, nil +} diff --git a/backend/internal/repository/instance_desired_state_repository.go b/backend/internal/repository/instance_desired_state_repository.go new file mode 100644 index 0000000..e90928d --- /dev/null +++ b/backend/internal/repository/instance_desired_state_repository.go @@ -0,0 +1,57 @@ +package repository + +import ( + "fmt" + "time" + + "clawreef/internal/models" + + "github.com/upper/db/v4" +) + +type InstanceDesiredStateRepository interface { + GetByInstanceID(instanceID int) (*models.InstanceDesiredState, error) + Create(state *models.InstanceDesiredState) error + Update(state *models.InstanceDesiredState) error +} + +type instanceDesiredStateRepository struct { + sess db.Session +} + +func NewInstanceDesiredStateRepository(sess db.Session) InstanceDesiredStateRepository { + return &instanceDesiredStateRepository{sess: sess} +} + +func (r *instanceDesiredStateRepository) GetByInstanceID(instanceID int) (*models.InstanceDesiredState, error) { + var item models.InstanceDesiredState + if err := r.sess.Collection("instance_desired_state").Find(db.Cond{"instance_id": instanceID}).One(&item); err != nil { + if err == db.ErrNoMoreRows { + return nil, nil + } + return nil, fmt.Errorf("failed to get instance desired state: %w", err) + } + return &item, nil +} + +func (r *instanceDesiredStateRepository) Create(state *models.InstanceDesiredState) error { + ensureTimestamps(&state.CreatedAt, &state.UpdatedAt) + res, err := r.sess.Collection("instance_desired_state").Insert(state) + if err != nil { + return fmt.Errorf("failed to create instance desired state: %w", err) + } + if id, ok := res.ID().(int64); ok { + state.ID = int(id) + } + return nil +} + +func (r *instanceDesiredStateRepository) Update(state *models.InstanceDesiredState) error { + if state.UpdatedAt.IsZero() { + state.UpdatedAt = time.Now().UTC() + } + if err := r.sess.Collection("instance_desired_state").Find(db.Cond{"id": state.ID}).Update(state); err != nil { + return fmt.Errorf("failed to update instance desired state: %w", err) + } + return nil +} diff --git a/backend/internal/repository/instance_repository.go b/backend/internal/repository/instance_repository.go index f486d6d..12bc667 100644 --- a/backend/internal/repository/instance_repository.go +++ b/backend/internal/repository/instance_repository.go @@ -13,6 +13,9 @@ type InstanceRepository interface { Create(instance *models.Instance) error GetByID(id int) (*models.Instance, error) GetByAccessToken(accessToken string) (*models.Instance, error) + GetByAgentBootstrapToken(bootstrapToken string) (*models.Instance, error) + GetAll(offset, limit int) ([]models.Instance, error) + CountAll() (int, error) GetByUserID(userID int, offset, limit int) ([]models.Instance, error) CountByUserID(userID int) (int, error) ExistsByUserIDAndName(userID int, name string) (bool, error) @@ -70,6 +73,35 @@ func (r *instanceRepository) GetByAccessToken(accessToken string) (*models.Insta return &instance, nil } +func (r *instanceRepository) GetByAgentBootstrapToken(bootstrapToken string) (*models.Instance, error) { + var instance models.Instance + err := r.sess.Collection("instances").Find(db.Cond{"agent_bootstrap_token": bootstrapToken}).One(&instance) + if err != nil { + if err == db.ErrNoMoreRows { + return nil, nil + } + return nil, fmt.Errorf("failed to get instance by agent bootstrap token: %w", err) + } + return &instance, nil +} + +func (r *instanceRepository) GetAll(offset, limit int) ([]models.Instance, error) { + var instances []models.Instance + err := r.sess.Collection("instances").Find().Offset(offset).Limit(limit).All(&instances) + if err != nil { + return nil, fmt.Errorf("failed to get all instances: %w", err) + } + return instances, nil +} + +func (r *instanceRepository) CountAll() (int, error) { + count, err := r.sess.Collection("instances").Find().Count() + if err != nil { + return 0, fmt.Errorf("failed to count all instances: %w", err) + } + return int(count), nil +} + // GetByUserID gets instances by user ID with pagination func (r *instanceRepository) GetByUserID(userID int, offset, limit int) ([]models.Instance, error) { var instances []models.Instance diff --git a/backend/internal/repository/instance_runtime_status_repository.go b/backend/internal/repository/instance_runtime_status_repository.go new file mode 100644 index 0000000..56d339a --- /dev/null +++ b/backend/internal/repository/instance_runtime_status_repository.go @@ -0,0 +1,57 @@ +package repository + +import ( + "fmt" + "time" + + "clawreef/internal/models" + + "github.com/upper/db/v4" +) + +type InstanceRuntimeStatusRepository interface { + GetByInstanceID(instanceID int) (*models.InstanceRuntimeStatus, error) + Create(status *models.InstanceRuntimeStatus) error + Update(status *models.InstanceRuntimeStatus) error +} + +type instanceRuntimeStatusRepository struct { + sess db.Session +} + +func NewInstanceRuntimeStatusRepository(sess db.Session) InstanceRuntimeStatusRepository { + return &instanceRuntimeStatusRepository{sess: sess} +} + +func (r *instanceRuntimeStatusRepository) GetByInstanceID(instanceID int) (*models.InstanceRuntimeStatus, error) { + var item models.InstanceRuntimeStatus + if err := r.sess.Collection("instance_runtime_status").Find(db.Cond{"instance_id": instanceID}).One(&item); err != nil { + if err == db.ErrNoMoreRows { + return nil, nil + } + return nil, fmt.Errorf("failed to get instance runtime status: %w", err) + } + return &item, nil +} + +func (r *instanceRuntimeStatusRepository) Create(status *models.InstanceRuntimeStatus) error { + ensureTimestamps(&status.CreatedAt, &status.UpdatedAt) + res, err := r.sess.Collection("instance_runtime_status").Insert(status) + if err != nil { + return fmt.Errorf("failed to create instance runtime status: %w", err) + } + if id, ok := res.ID().(int64); ok { + status.ID = int(id) + } + return nil +} + +func (r *instanceRuntimeStatusRepository) Update(status *models.InstanceRuntimeStatus) error { + if status.UpdatedAt.IsZero() { + status.UpdatedAt = time.Now().UTC() + } + if err := r.sess.Collection("instance_runtime_status").Find(db.Cond{"id": status.ID}).Update(status); err != nil { + return fmt.Errorf("failed to update instance runtime status: %w", err) + } + return nil +} diff --git a/backend/internal/repository/timestamps.go b/backend/internal/repository/timestamps.go new file mode 100644 index 0000000..5d9da59 --- /dev/null +++ b/backend/internal/repository/timestamps.go @@ -0,0 +1,13 @@ +package repository + +import "time" + +func ensureTimestamps(createdAt *time.Time, updatedAt *time.Time) { + now := time.Now().UTC() + if createdAt != nil && createdAt.IsZero() { + *createdAt = now + } + if updatedAt != nil && updatedAt.IsZero() { + *updatedAt = now + } +} diff --git a/backend/internal/services/instance_agent_service.go b/backend/internal/services/instance_agent_service.go new file mode 100644 index 0000000..adc29af --- /dev/null +++ b/backend/internal/services/instance_agent_service.go @@ -0,0 +1,414 @@ +package services + +import ( + "crypto/rand" + "encoding/hex" + "encoding/json" + "fmt" + "strings" + "time" + + "clawreef/internal/models" + "clawreef/internal/repository" +) + +const ( + AgentProtocolVersionV1 = "v1" + + agentStatusOnline = "online" + agentStatusOffline = "offline" + agentStatusStale = "stale" + + openClawStatusUnknown = "unknown" +) + +type AgentRegisterRequest struct { + InstanceID int `json:"instance_id" binding:"required,min=1"` + AgentID string `json:"agent_id" binding:"required"` + AgentVersion string `json:"agent_version" binding:"required"` + ProtocolVersion string `json:"protocol_version" binding:"required"` + Capabilities []string `json:"capabilities"` + HostInfo map[string]interface{} `json:"host_info"` +} + +type AgentRegisterResponse struct { + SessionToken string `json:"session_token"` + SessionExpiresAt time.Time `json:"session_expires_at"` + HeartbeatIntervalSeconds int `json:"heartbeat_interval_seconds"` + CommandPollIntervalSeconds int `json:"command_poll_interval_seconds"` + ServerTime time.Time `json:"server_time"` +} + +type AgentHeartbeatRequest struct { + AgentID string `json:"agent_id" binding:"required"` + Timestamp time.Time `json:"timestamp"` + OpenClawStatus string `json:"openclaw_status"` + CurrentConfigRevisionID *int `json:"current_config_revision_id,omitempty"` + Summary map[string]interface{} `json:"summary"` +} + +type AgentHeartbeatResponse struct { + ServerTime time.Time `json:"server_time"` + HasPendingCommand bool `json:"has_pending_command"` + DesiredPowerState string `json:"desired_power_state"` + DesiredConfigRevisionID *int `json:"desired_config_revision_id,omitempty"` +} + +type InstanceAgentPayload struct { + AgentID string `json:"agent_id"` + AgentVersion string `json:"agent_version"` + ProtocolVersion string `json:"protocol_version"` + Status string `json:"status"` + Capabilities []string `json:"capabilities"` + HostInfo map[string]interface{} `json:"host_info,omitempty"` + LastHeartbeatAt *time.Time `json:"last_heartbeat_at,omitempty"` + LastReportedAt *time.Time `json:"last_reported_at,omitempty"` + LastSeenIP *string `json:"last_seen_ip,omitempty"` + RegisteredAt *time.Time `json:"registered_at,omitempty"` +} + +type AgentSession struct { + Instance *models.Instance + Agent *models.InstanceAgent +} + +type InstanceAgentService interface { + Register(bootstrapToken string, req AgentRegisterRequest, clientIP string) (*AgentRegisterResponse, error) + AuthenticateSession(sessionToken string) (*AgentSession, error) + Heartbeat(session *AgentSession, req AgentHeartbeatRequest, clientIP string) (*AgentHeartbeatResponse, error) + GetPayloadByInstanceID(instanceID int) (*InstanceAgentPayload, error) +} + +type instanceAgentService struct { + instanceRepo repository.InstanceRepository + agentRepo repository.InstanceAgentRepository + desiredStateRepo repository.InstanceDesiredStateRepository + runtimeRepo repository.InstanceRuntimeStatusRepository + commandRepo repository.InstanceCommandRepository +} + +func NewInstanceAgentService(instanceRepo repository.InstanceRepository, agentRepo repository.InstanceAgentRepository, desiredStateRepo repository.InstanceDesiredStateRepository, runtimeRepo repository.InstanceRuntimeStatusRepository, commandRepo repository.InstanceCommandRepository) InstanceAgentService { + return &instanceAgentService{ + instanceRepo: instanceRepo, + agentRepo: agentRepo, + desiredStateRepo: desiredStateRepo, + runtimeRepo: runtimeRepo, + commandRepo: commandRepo, + } +} + +func (s *instanceAgentService) Register(bootstrapToken string, req AgentRegisterRequest, clientIP string) (*AgentRegisterResponse, error) { + bootstrapToken = strings.TrimSpace(bootstrapToken) + if bootstrapToken == "" { + return nil, fmt.Errorf("agent bootstrap token is required") + } + if strings.TrimSpace(req.AgentID) == "" { + return nil, fmt.Errorf("agent id is required") + } + if strings.TrimSpace(req.ProtocolVersion) != AgentProtocolVersionV1 { + return nil, fmt.Errorf("unsupported agent protocol version") + } + + instance, err := s.instanceRepo.GetByAgentBootstrapToken(bootstrapToken) + if err != nil { + return nil, err + } + if instance == nil || instance.ID != req.InstanceID { + return nil, fmt.Errorf("invalid agent bootstrap token") + } + if !strings.EqualFold(instance.Type, "openclaw") { + return nil, fmt.Errorf("agent registration is only supported for openclaw instances") + } + + now := time.Now().UTC() + sessionToken, err := generatePrefixedToken("agt_sess") + if err != nil { + return nil, fmt.Errorf("failed to generate agent session token: %w", err) + } + sessionExpiresAt := now.Add(24 * time.Hour) + + capabilitiesJSON, err := marshalJSON(req.Capabilities) + if err != nil { + return nil, fmt.Errorf("failed to encode agent capabilities: %w", err) + } + hostInfoJSON, err := marshalOptionalJSON(req.HostInfo) + if err != nil { + return nil, fmt.Errorf("failed to encode agent host info: %w", err) + } + + agent, err := s.agentRepo.GetByInstanceID(instance.ID) + if err != nil { + return nil, err + } + if agent == nil { + agent = &models.InstanceAgent{ + InstanceID: instance.ID, + AgentID: strings.TrimSpace(req.AgentID), + AgentVersion: strings.TrimSpace(req.AgentVersion), + ProtocolVersion: strings.TrimSpace(req.ProtocolVersion), + Status: agentStatusOnline, + CapabilitiesJSON: capabilitiesJSON, + HostInfoJSON: hostInfoJSON, + SessionToken: &sessionToken, + SessionExpiresAt: &sessionExpiresAt, + LastHeartbeatAt: &now, + LastReportedAt: nil, + LastSeenIP: optionalString(strings.TrimSpace(clientIP)), + RegisteredAt: &now, + } + if err := s.agentRepo.Create(agent); err != nil { + return nil, err + } + } else { + agent.AgentID = strings.TrimSpace(req.AgentID) + agent.AgentVersion = strings.TrimSpace(req.AgentVersion) + agent.ProtocolVersion = strings.TrimSpace(req.ProtocolVersion) + agent.Status = agentStatusOnline + agent.CapabilitiesJSON = capabilitiesJSON + agent.HostInfoJSON = hostInfoJSON + agent.SessionToken = &sessionToken + agent.SessionExpiresAt = &sessionExpiresAt + agent.LastHeartbeatAt = &now + agent.LastSeenIP = optionalString(strings.TrimSpace(clientIP)) + agent.RegisteredAt = &now + if err := s.agentRepo.Update(agent); err != nil { + return nil, err + } + } + + if _, err := s.ensureDesiredState(instance); err != nil { + return nil, err + } + if _, err := s.ensureRuntimeStatus(instance.ID); err != nil { + return nil, err + } + + return &AgentRegisterResponse{ + SessionToken: sessionToken, + SessionExpiresAt: sessionExpiresAt, + HeartbeatIntervalSeconds: 15, + CommandPollIntervalSeconds: 5, + ServerTime: now, + }, nil +} + +func (s *instanceAgentService) AuthenticateSession(sessionToken string) (*AgentSession, error) { + sessionToken = strings.TrimSpace(sessionToken) + if sessionToken == "" { + return nil, fmt.Errorf("agent session token is required") + } + + agent, err := s.agentRepo.GetBySessionToken(sessionToken) + if err != nil { + return nil, err + } + if agent == nil || agent.SessionExpiresAt == nil || agent.SessionExpiresAt.Before(time.Now().UTC()) { + return nil, fmt.Errorf("invalid or expired agent session token") + } + + instance, err := s.instanceRepo.GetByID(agent.InstanceID) + if err != nil { + return nil, err + } + if instance == nil { + return nil, fmt.Errorf("instance not found") + } + + return &AgentSession{Instance: instance, Agent: agent}, nil +} + +func (s *instanceAgentService) Heartbeat(session *AgentSession, req AgentHeartbeatRequest, clientIP string) (*AgentHeartbeatResponse, error) { + if session == nil || session.Agent == nil || session.Instance == nil { + return nil, fmt.Errorf("agent session is required") + } + if strings.TrimSpace(req.AgentID) == "" || strings.TrimSpace(req.AgentID) != session.Agent.AgentID { + return nil, fmt.Errorf("agent id does not match session") + } + + now := time.Now().UTC() + session.Agent.Status = agentStatusOnline + session.Agent.LastHeartbeatAt = &now + session.Agent.LastSeenIP = optionalString(strings.TrimSpace(clientIP)) + if err := s.agentRepo.Update(session.Agent); err != nil { + return nil, err + } + + runtimeStatus, err := s.ensureRuntimeStatus(session.Instance.ID) + if err != nil { + return nil, err + } + runtimeStatus.AgentStatus = agentStatusOnline + if strings.TrimSpace(req.OpenClawStatus) != "" { + runtimeStatus.OpenClawStatus = strings.TrimSpace(req.OpenClawStatus) + } + runtimeStatus.CurrentConfigRevisionID = req.CurrentConfigRevisionID + if req.Summary != nil { + summaryJSON, err := marshalOptionalJSON(req.Summary) + if err != nil { + return nil, fmt.Errorf("failed to encode heartbeat summary: %w", err) + } + runtimeStatus.SummaryJSON = summaryJSON + if openClawPID, ok := req.Summary["openclaw_pid"].(float64); ok { + pid := int(openClawPID) + runtimeStatus.OpenClawPID = &pid + } + } + runtimeStatus.LastReportedAt = &now + if err := s.runtimeRepo.Update(runtimeStatus); err != nil { + return nil, err + } + + desiredState, err := s.ensureDesiredState(session.Instance) + if err != nil { + return nil, err + } + nextCommand, err := s.commandRepo.GetNextPendingByInstance(session.Instance.ID) + if err != nil { + return nil, err + } + + return &AgentHeartbeatResponse{ + ServerTime: now, + HasPendingCommand: nextCommand != nil, + DesiredPowerState: desiredState.DesiredPowerState, + DesiredConfigRevisionID: desiredState.DesiredConfigRevisionID, + }, nil +} + +func (s *instanceAgentService) GetPayloadByInstanceID(instanceID int) (*InstanceAgentPayload, error) { + agent, err := s.agentRepo.GetByInstanceID(instanceID) + if err != nil { + return nil, err + } + if agent == nil { + return nil, nil + } + + payload := &InstanceAgentPayload{ + AgentID: agent.AgentID, + AgentVersion: agent.AgentVersion, + ProtocolVersion: agent.ProtocolVersion, + Status: deriveAgentStatus(agent), + LastHeartbeatAt: agent.LastHeartbeatAt, + LastReportedAt: agent.LastReportedAt, + LastSeenIP: agent.LastSeenIP, + RegisteredAt: agent.RegisteredAt, + } + if err := unmarshalJSON(agent.CapabilitiesJSON, &payload.Capabilities); err != nil { + return nil, fmt.Errorf("failed to decode agent capabilities: %w", err) + } + if agent.HostInfoJSON != nil && strings.TrimSpace(*agent.HostInfoJSON) != "" { + if err := unmarshalJSON(*agent.HostInfoJSON, &payload.HostInfo); err != nil { + return nil, fmt.Errorf("failed to decode agent host info: %w", err) + } + } + return payload, nil +} + +func (s *instanceAgentService) ensureDesiredState(instance *models.Instance) (*models.InstanceDesiredState, error) { + state, err := s.desiredStateRepo.GetByInstanceID(instance.ID) + if err != nil { + return nil, err + } + if state != nil { + return state, nil + } + + desiredPowerState := "stopped" + if instance.Status == "running" || instance.Status == "creating" { + desiredPowerState = "running" + } + now := time.Now().UTC() + state = &models.InstanceDesiredState{ + InstanceID: instance.ID, + DesiredPowerState: desiredPowerState, + UpdatedAt: now, + CreatedAt: now, + } + if err := s.desiredStateRepo.Create(state); err != nil { + return nil, err + } + return state, nil +} + +func (s *instanceAgentService) ensureRuntimeStatus(instanceID int) (*models.InstanceRuntimeStatus, error) { + status, err := s.runtimeRepo.GetByInstanceID(instanceID) + if err != nil { + return nil, err + } + if status != nil { + return status, nil + } + + now := time.Now().UTC() + status = &models.InstanceRuntimeStatus{ + InstanceID: instanceID, + InfraStatus: "creating", + AgentStatus: agentStatusOffline, + OpenClawStatus: openClawStatusUnknown, + CreatedAt: now, + UpdatedAt: now, + } + if err := s.runtimeRepo.Create(status); err != nil { + return nil, err + } + return status, nil +} + +func deriveAgentStatus(agent *models.InstanceAgent) string { + if agent == nil || agent.LastHeartbeatAt == nil { + return agentStatusOffline + } + age := time.Since(*agent.LastHeartbeatAt) + switch { + case age <= 45*time.Second: + return agentStatusOnline + case age <= 120*time.Second: + return agentStatusStale + default: + return agentStatusOffline + } +} + +func generatePrefixedToken(prefix string) (string, error) { + bytes := make([]byte, 24) + if _, err := rand.Read(bytes); err != nil { + return "", err + } + return prefix + "_" + hex.EncodeToString(bytes), nil +} + +func marshalJSON(value interface{}) (string, error) { + raw, err := json.Marshal(value) + if err != nil { + return "", err + } + return string(raw), nil +} + +func marshalOptionalJSON(value interface{}) (*string, error) { + if value == nil { + return nil, nil + } + raw, err := json.Marshal(value) + if err != nil { + return nil, err + } + encoded := string(raw) + return &encoded, nil +} + +func unmarshalJSON(raw string, target interface{}) error { + if strings.TrimSpace(raw) == "" { + return nil + } + return json.Unmarshal([]byte(raw), target) +} + +func optionalString(value string) *string { + if strings.TrimSpace(value) == "" { + return nil + } + trimmed := strings.TrimSpace(value) + return &trimmed +} diff --git a/backend/internal/services/instance_command_service.go b/backend/internal/services/instance_command_service.go new file mode 100644 index 0000000..69c6056 --- /dev/null +++ b/backend/internal/services/instance_command_service.go @@ -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 +} diff --git a/backend/internal/services/instance_config_revision_service.go b/backend/internal/services/instance_config_revision_service.go new file mode 100644 index 0000000..6e2d0de --- /dev/null +++ b/backend/internal/services/instance_config_revision_service.go @@ -0,0 +1,124 @@ +package services + +import ( + "crypto/sha256" + "encoding/hex" + "encoding/json" + "fmt" + "strings" + "time" + + "clawreef/internal/models" + "clawreef/internal/repository" +) + +type InstanceConfigRevisionPayload struct { + ID int `json:"id"` + InstanceID int `json:"instance_id"` + SourceSnapshotID *int `json:"source_snapshot_id,omitempty"` + SourceBundleID *int `json:"source_bundle_id,omitempty"` + RevisionNo int `json:"revision_no"` + Content json.RawMessage `json:"content"` + Checksum string `json:"checksum"` + Status string `json:"status"` + PublishedBy *int `json:"published_by,omitempty"` + PublishedAt *time.Time `json:"published_at,omitempty"` + ActivatedAt *time.Time `json:"activated_at,omitempty"` +} + +type InstanceConfigRevisionService interface { + GetByID(id int) (*InstanceConfigRevisionPayload, error) + ListByInstanceID(instanceID int, limit int) ([]InstanceConfigRevisionPayload, error) + CreateFromSnapshot(instanceID int, snapshot *models.OpenClawInjectionSnapshot, publishedBy *int) (*InstanceConfigRevisionPayload, error) +} + +type instanceConfigRevisionService struct { + repo repository.InstanceConfigRevisionRepository +} + +func NewInstanceConfigRevisionService(repo repository.InstanceConfigRevisionRepository) InstanceConfigRevisionService { + return &instanceConfigRevisionService{repo: repo} +} + +func (s *instanceConfigRevisionService) GetByID(id int) (*InstanceConfigRevisionPayload, error) { + item, err := s.repo.GetByID(id) + if err != nil { + return nil, err + } + if item == nil { + return nil, fmt.Errorf("instance config revision not found") + } + return configRevisionPayloadFromModel(*item) +} + +func (s *instanceConfigRevisionService) ListByInstanceID(instanceID int, limit int) ([]InstanceConfigRevisionPayload, error) { + items, err := s.repo.ListByInstanceID(instanceID, limit) + if err != nil { + return nil, err + } + + result := make([]InstanceConfigRevisionPayload, 0, len(items)) + for _, item := range items { + payload, err := configRevisionPayloadFromModel(item) + if err != nil { + return nil, err + } + result = append(result, *payload) + } + return result, nil +} + +func (s *instanceConfigRevisionService) CreateFromSnapshot(instanceID int, snapshot *models.OpenClawInjectionSnapshot, publishedBy *int) (*InstanceConfigRevisionPayload, error) { + if snapshot == nil { + return nil, fmt.Errorf("openclaw injection snapshot is required") + } + + latest, err := s.repo.GetLatestByInstanceID(instanceID) + if err != nil { + return nil, err + } + revisionNo := 1 + if latest != nil { + revisionNo = latest.RevisionNo + 1 + } + contentJSON := strings.TrimSpace(snapshot.RenderedManifestJSON) + if contentJSON == "" { + contentJSON = "{}" + } + sum := sha256.Sum256([]byte(contentJSON)) + checksum := "sha256:" + hex.EncodeToString(sum[:]) + now := time.Now().UTC() + revision := &models.InstanceConfigRevision{ + InstanceID: instanceID, + SourceSnapshotID: &snapshot.ID, + RevisionNo: revisionNo, + ContentJSON: contentJSON, + Checksum: checksum, + Status: "published", + PublishedBy: publishedBy, + PublishedAt: &now, + CreatedAt: now, + UpdatedAt: now, + } + if err := s.repo.Create(revision); err != nil { + return nil, err + } + return configRevisionPayloadFromModel(*revision) +} + +func configRevisionPayloadFromModel(item models.InstanceConfigRevision) (*InstanceConfigRevisionPayload, error) { + payload := &InstanceConfigRevisionPayload{ + ID: item.ID, + InstanceID: item.InstanceID, + SourceSnapshotID: item.SourceSnapshotID, + SourceBundleID: item.SourceBundleID, + RevisionNo: item.RevisionNo, + Checksum: item.Checksum, + Status: item.Status, + PublishedBy: item.PublishedBy, + PublishedAt: item.PublishedAt, + ActivatedAt: item.ActivatedAt, + Content: json.RawMessage(item.ContentJSON), + } + return payload, nil +} diff --git a/backend/internal/services/instance_runtime.go b/backend/internal/services/instance_runtime.go index b3eda68..48222b0 100644 --- a/backend/internal/services/instance_runtime.go +++ b/backend/internal/services/instance_runtime.go @@ -58,6 +58,7 @@ func buildRuntimeConfig(instanceType, osType, osVersion string, registry, tag *s "SUBFOLDER": "/", } case "openclaw": + config.MountPath = "/config" if (registry == nil || strings.TrimSpace(*registry) == "") && (tag == nil || strings.TrimSpace(*tag) == "") { config.Image = defaultSystemImageSettings["openclaw"] } else { @@ -84,7 +85,7 @@ func defaultPortForInstanceType(instanceType string) int32 { func defaultMountPathForInstanceType(instanceType string) string { switch instanceType { - case "ubuntu", "webtop": + case "ubuntu", "webtop", "openclaw": return "/config" default: return "/home/user/data" @@ -211,6 +212,46 @@ func defaultGatewayBaseURL() (string, bool) { return fmt.Sprintf("http://%s.%s.svc.cluster.local:%s/api/v1/gateway/llm", serviceName, systemNamespace, port), true } +func defaultAgentControlBaseURL() (string, bool) { + if override := strings.TrimSpace(os.Getenv("CLAWMANAGER_AGENT_CONTROL_BASE_URL")); override != "" { + return override, true + } + + systemNamespace := strings.TrimSpace(os.Getenv("CLAWMANAGER_SYSTEM_NAMESPACE")) + if systemNamespace == "" { + if client := k8s.GetClient(); client != nil { + systemNamespace = client.GetSystemNamespace() + } else if baseNamespace := strings.TrimSpace(os.Getenv("K8S_NAMESPACE")); baseNamespace != "" { + systemNamespace = fmt.Sprintf("%s-system", baseNamespace) + } + } + if systemNamespace == "" { + return "", false + } + + serviceName := strings.TrimSpace(os.Getenv("CLAWMANAGER_AGENT_CONTROL_SERVICE_NAME")) + if serviceName == "" { + serviceName = strings.TrimSpace(os.Getenv("CLAWMANAGER_LLM_GATEWAY_SERVICE_NAME")) + } + if serviceName == "" { + serviceName = strings.TrimSpace(os.Getenv("CLAWMANAGER_LLM_GATEWAY_SERVICE")) + } + if serviceName == "" { + serviceName = "clawmanager-gateway" + } + + port := normalizePortValue( + strings.TrimSpace(os.Getenv("CLAWMANAGER_AGENT_CONTROL_SERVICE_PORT")), + strings.TrimSpace(os.Getenv("CLAWMANAGER_LLM_GATEWAY_SERVICE_PORT")), + strings.TrimSpace(os.Getenv("CLAWMANAGER_LLM_GATEWAY_PORT")), + ) + if port == "" { + port = "9001" + } + + return fmt.Sprintf("http://%s.%s.svc.cluster.local:%s", serviceName, systemNamespace, port), true +} + func defaultNoProxyList() string { if override := strings.TrimSpace(os.Getenv("CLAWMANAGER_NO_PROXY")); override != "" { return override diff --git a/backend/internal/services/instance_runtime_status_service.go b/backend/internal/services/instance_runtime_status_service.go new file mode 100644 index 0000000..25964ac --- /dev/null +++ b/backend/internal/services/instance_runtime_status_service.go @@ -0,0 +1,190 @@ +package services + +import ( + "encoding/json" + "fmt" + "strings" + "time" + + "clawreef/internal/models" + "clawreef/internal/repository" +) + +type AgentStateReportRequest struct { + AgentID string `json:"agent_id" binding:"required"` + ReportedAt *time.Time `json:"reported_at,omitempty"` + Runtime AgentRuntimePayload `json:"runtime"` + SystemInfo map[string]interface{} `json:"system_info"` + Health map[string]interface{} `json:"health"` +} + +type AgentRuntimePayload struct { + OpenClawStatus string `json:"openclaw_status"` + OpenClawPID *int `json:"openclaw_pid,omitempty"` + OpenClawVersion string `json:"openclaw_version"` + CurrentConfigRevisionID *int `json:"current_config_revision_id,omitempty"` +} + +type InstanceRuntimeStatusPayload struct { + InstanceID int `json:"instance_id"` + InfraStatus string `json:"infra_status"` + AgentStatus string `json:"agent_status"` + OpenClawStatus string `json:"openclaw_status"` + OpenClawPID *int `json:"openclaw_pid,omitempty"` + OpenClawVersion *string `json:"openclaw_version,omitempty"` + CurrentConfigRevisionID *int `json:"current_config_revision_id,omitempty"` + DesiredConfigRevisionID *int `json:"desired_config_revision_id,omitempty"` + SystemInfo map[string]interface{} `json:"system_info,omitempty"` + Health map[string]interface{} `json:"health,omitempty"` + Summary map[string]interface{} `json:"summary,omitempty"` + LastReportedAt *time.Time `json:"last_reported_at,omitempty"` +} + +type InstanceRuntimeStatusService interface { + Report(session *AgentSession, req AgentStateReportRequest, clientIP string) error + GetByInstanceID(instanceID int) (*InstanceRuntimeStatusPayload, error) + UpsertInfraStatus(instanceID int, infraStatus string) error +} + +type instanceRuntimeStatusService struct { + runtimeRepo repository.InstanceRuntimeStatusRepository + agentRepo repository.InstanceAgentRepository + desiredStateRepo repository.InstanceDesiredStateRepository +} + +func NewInstanceRuntimeStatusService(runtimeRepo repository.InstanceRuntimeStatusRepository, agentRepo repository.InstanceAgentRepository, desiredStateRepo repository.InstanceDesiredStateRepository) InstanceRuntimeStatusService { + return &instanceRuntimeStatusService{ + runtimeRepo: runtimeRepo, + agentRepo: agentRepo, + desiredStateRepo: desiredStateRepo, + } +} + +func (s *instanceRuntimeStatusService) Report(session *AgentSession, req AgentStateReportRequest, clientIP string) error { + if session == nil || session.Agent == nil || session.Instance == nil { + return fmt.Errorf("agent session is required") + } + if strings.TrimSpace(req.AgentID) != session.Agent.AgentID { + return fmt.Errorf("agent id does not match session") + } + + status, err := s.getOrCreate(session.Instance.ID) + if err != nil { + return err + } + now := time.Now().UTC() + reportedAt := req.ReportedAt + if reportedAt == nil || reportedAt.IsZero() { + reportedAt = &now + } + + status.AgentStatus = agentStatusOnline + if strings.TrimSpace(req.Runtime.OpenClawStatus) != "" { + status.OpenClawStatus = strings.TrimSpace(req.Runtime.OpenClawStatus) + } + status.OpenClawPID = req.Runtime.OpenClawPID + if strings.TrimSpace(req.Runtime.OpenClawVersion) != "" { + version := strings.TrimSpace(req.Runtime.OpenClawVersion) + status.OpenClawVersion = &version + } + status.CurrentConfigRevisionID = req.Runtime.CurrentConfigRevisionID + status.LastReportedAt = reportedAt + + systemInfoJSON, err := marshalOptionalJSON(req.SystemInfo) + if err != nil { + return fmt.Errorf("failed to encode system info: %w", err) + } + healthJSON, err := marshalOptionalJSON(req.Health) + if err != nil { + return fmt.Errorf("failed to encode health info: %w", err) + } + status.SystemInfoJSON = systemInfoJSON + status.HealthJSON = healthJSON + if err := s.runtimeRepo.Update(status); err != nil { + return err + } + + session.Agent.Status = agentStatusOnline + session.Agent.LastHeartbeatAt = reportedAt + session.Agent.LastReportedAt = reportedAt + session.Agent.LastSeenIP = optionalString(strings.TrimSpace(clientIP)) + if err := s.agentRepo.Update(session.Agent); err != nil { + return err + } + + return nil +} + +func (s *instanceRuntimeStatusService) GetByInstanceID(instanceID int) (*InstanceRuntimeStatusPayload, error) { + status, err := s.runtimeRepo.GetByInstanceID(instanceID) + if err != nil { + return nil, err + } + if status == nil { + return nil, nil + } + payload := &InstanceRuntimeStatusPayload{ + InstanceID: status.InstanceID, + InfraStatus: status.InfraStatus, + AgentStatus: status.AgentStatus, + OpenClawStatus: status.OpenClawStatus, + OpenClawPID: status.OpenClawPID, + OpenClawVersion: status.OpenClawVersion, + CurrentConfigRevisionID: status.CurrentConfigRevisionID, + DesiredConfigRevisionID: status.DesiredConfigRevisionID, + LastReportedAt: status.LastReportedAt, + } + if status.SystemInfoJSON != nil && strings.TrimSpace(*status.SystemInfoJSON) != "" { + if err := json.Unmarshal([]byte(*status.SystemInfoJSON), &payload.SystemInfo); err != nil { + return nil, fmt.Errorf("failed to decode system info: %w", err) + } + } + if status.HealthJSON != nil && strings.TrimSpace(*status.HealthJSON) != "" { + if err := json.Unmarshal([]byte(*status.HealthJSON), &payload.Health); err != nil { + return nil, fmt.Errorf("failed to decode health info: %w", err) + } + } + if status.SummaryJSON != nil && strings.TrimSpace(*status.SummaryJSON) != "" { + if err := json.Unmarshal([]byte(*status.SummaryJSON), &payload.Summary); err != nil { + return nil, fmt.Errorf("failed to decode runtime summary: %w", err) + } + } + return payload, nil +} + +func (s *instanceRuntimeStatusService) UpsertInfraStatus(instanceID int, infraStatus string) error { + status, err := s.getOrCreate(instanceID) + if err != nil { + return err + } + status.InfraStatus = infraStatus + + desiredState, err := s.desiredStateRepo.GetByInstanceID(instanceID) + if err == nil && desiredState != nil { + status.DesiredConfigRevisionID = desiredState.DesiredConfigRevisionID + } + return s.runtimeRepo.Update(status) +} + +func (s *instanceRuntimeStatusService) getOrCreate(instanceID int) (*models.InstanceRuntimeStatus, error) { + status, err := s.runtimeRepo.GetByInstanceID(instanceID) + if err != nil { + return nil, err + } + if status != nil { + return status, nil + } + now := time.Now().UTC() + status = &models.InstanceRuntimeStatus{ + InstanceID: instanceID, + InfraStatus: "creating", + AgentStatus: agentStatusOffline, + OpenClawStatus: openClawStatusUnknown, + CreatedAt: now, + UpdatedAt: now, + } + if err := s.runtimeRepo.Create(status); err != nil { + return nil, err + } + return status, nil +} diff --git a/backend/internal/services/instance_service.go b/backend/internal/services/instance_service.go index e47dede..427eb2b 100644 --- a/backend/internal/services/instance_service.go +++ b/backend/internal/services/instance_service.go @@ -5,6 +5,7 @@ import ( "crypto/rand" "encoding/hex" "fmt" + "strconv" "strings" "time" @@ -20,6 +21,7 @@ type InstanceService interface { Create(userID int, req CreateInstanceRequest) (*models.Instance, error) GetByID(id int) (*models.Instance, error) GetByUserID(userID int, offset, limit int) ([]models.Instance, int, error) + GetVisibleInstances(userID int, userRole string, offset, limit int) ([]models.Instance, int, error) Start(instanceID int) error Stop(instanceID int) error Restart(instanceID int) error @@ -210,12 +212,21 @@ func (s *instanceService) Create(userID int, req CreateInstanceRequest) (*models s.instanceRepo.Delete(instance.ID) return nil, fmt.Errorf("failed to provision instance gateway token: %w", err) } + if _, err := s.ensureAgentBootstrapToken(instance); err != nil { + s.instanceRepo.Delete(instance.ID) + return nil, fmt.Errorf("failed to provision instance agent bootstrap token: %w", err) + } gatewayEnv, err := s.buildGatewayEnv(instance) if err != nil { s.instanceRepo.Delete(instance.ID) return nil, fmt.Errorf("failed to build instance gateway config: %w", err) } + agentEnv, err := s.buildAgentEnv(instance) + if err != nil { + s.instanceRepo.Delete(instance.ID) + return nil, fmt.Errorf("failed to build instance agent config: %w", err) + } var bootstrapSnapshot *models.OpenClawInjectionSnapshot var bootstrapSecretName string @@ -281,7 +292,7 @@ func (s *instanceService) Create(userID int, req CreateInstanceRequest) (*models Image: runtimeConfig.Image, MountPath: runtimeConfig.MountPath, ContainerPort: runtimeConfig.Port, - ExtraEnv: withInstanceProxyEnv(instance.Type, instance.ID, mergeEnvMaps(runtimeConfig.Env, gatewayEnv)), + ExtraEnv: withInstanceProxyEnv(instance.Type, instance.ID, mergeEnvMaps(runtimeConfig.Env, mergeEnvMaps(gatewayEnv, agentEnv))), EnvFromSecretNames: []string{bootstrapSecretName}, } @@ -377,6 +388,24 @@ func (s *instanceService) GetByUserID(userID int, offset, limit int) ([]models.I return instances, total, nil } +func (s *instanceService) GetVisibleInstances(userID int, userRole string, offset, limit int) ([]models.Instance, int, error) { + if strings.EqualFold(strings.TrimSpace(userRole), "admin") { + instances, err := s.instanceRepo.GetAll(offset, limit) + if err != nil { + return nil, 0, err + } + + total, err := s.instanceRepo.CountAll() + if err != nil { + return nil, 0, err + } + + return instances, total, nil + } + + return s.GetByUserID(userID, offset, limit) +} + // Start starts an instance func (s *instanceService) Start(instanceID int) error { ctx := context.Background() @@ -397,11 +426,18 @@ func (s *instanceService) Start(instanceID int) error { if _, err := s.ensureGatewayToken(instance); err != nil { return fmt.Errorf("failed to provision instance gateway token: %w", err) } + if _, err := s.ensureAgentBootstrapToken(instance); err != nil { + return fmt.Errorf("failed to provision instance agent bootstrap token: %w", err) + } gatewayEnv, err := s.buildGatewayEnv(instance) if err != nil { return fmt.Errorf("failed to build instance gateway config: %w", err) } + agentEnv, err := s.buildAgentEnv(instance) + if err != nil { + return fmt.Errorf("failed to build instance agent config: %w", err) + } bootstrapSecretName := "" if strings.EqualFold(instance.Type, "openclaw") && s.openClawConfigService != nil && instance.OpenClawConfigSnapshotID != nil && *instance.OpenClawConfigSnapshotID > 0 { @@ -431,7 +467,7 @@ func (s *instanceService) Start(instanceID int) error { Image: runtimeConfig.Image, MountPath: instance.MountPath, ContainerPort: runtimeConfig.Port, - ExtraEnv: withInstanceProxyEnv(instance.Type, instance.ID, mergeEnvMaps(runtimeConfig.Env, gatewayEnv)), + ExtraEnv: withInstanceProxyEnv(instance.Type, instance.ID, mergeEnvMaps(runtimeConfig.Env, mergeEnvMaps(gatewayEnv, agentEnv))), EnvFromSecretNames: []string{bootstrapSecretName}, } @@ -533,6 +569,49 @@ func (s *instanceService) buildGatewayEnv(instance *models.Instance) (map[string }, nil } +func (s *instanceService) ensureAgentBootstrapToken(instance *models.Instance) (string, error) { + if instance.AgentBootstrapToken != nil && strings.TrimSpace(*instance.AgentBootstrapToken) != "" { + return strings.TrimSpace(*instance.AgentBootstrapToken), nil + } + + token, err := generatePrefixedToken("agt_boot") + if err != nil { + return "", fmt.Errorf("failed to generate instance agent bootstrap token: %w", err) + } + instance.AgentBootstrapToken = &token + instance.UpdatedAt = time.Now() + if err := s.instanceRepo.Update(instance); err != nil { + return "", fmt.Errorf("failed to persist instance agent bootstrap token: %w", err) + } + return token, nil +} + +func (s *instanceService) buildAgentEnv(instance *models.Instance) (map[string]string, error) { + if instance == nil || !strings.EqualFold(strings.TrimSpace(instance.Type), "openclaw") { + return map[string]string{}, nil + } + if instance.AgentBootstrapToken == nil || strings.TrimSpace(*instance.AgentBootstrapToken) == "" { + return nil, fmt.Errorf("instance agent bootstrap token is not configured") + } + + baseURL, ok := defaultAgentControlBaseURL() + if !ok { + return nil, fmt.Errorf("agent control base URL is not configured") + } + + diskLimitBytes := int64(instance.DiskGB) * 1024 * 1024 * 1024 + + return map[string]string{ + "CLAWMANAGER_AGENT_ENABLED": "true", + "CLAWMANAGER_AGENT_BASE_URL": baseURL, + "CLAWMANAGER_AGENT_BOOTSTRAP_TOKEN": strings.TrimSpace(*instance.AgentBootstrapToken), + "CLAWMANAGER_AGENT_DISK_LIMIT_BYTES": strconv.FormatInt(diskLimitBytes, 10), + "CLAWMANAGER_AGENT_INSTANCE_ID": fmt.Sprintf("%d", instance.ID), + "CLAWMANAGER_AGENT_PERSISTENT_DIR": "/config", + "CLAWMANAGER_AGENT_PROTOCOL_VERSION": AgentProtocolVersionV1, + }, nil +} + func (s *instanceService) resolveDefaultGatewayModel() (string, error) { if s.llmModelRepo == nil { return "", fmt.Errorf("llm model repository not configured") diff --git a/backend/internal/services/sync_service.go b/backend/internal/services/sync_service.go index 88ac38b..7268c2b 100644 --- a/backend/internal/services/sync_service.go +++ b/backend/internal/services/sync_service.go @@ -14,19 +14,21 @@ import ( // SyncService handles synchronization between database and K8s state type SyncService struct { - instanceRepo repository.InstanceRepository - podService *k8s.PodService - interval time.Duration - stopChan chan struct{} + instanceRepo repository.InstanceRepository + runtimeStatusService InstanceRuntimeStatusService + podService *k8s.PodService + interval time.Duration + stopChan chan struct{} } // NewSyncService creates a new sync service -func NewSyncService(instanceRepo repository.InstanceRepository) *SyncService { +func NewSyncService(instanceRepo repository.InstanceRepository, runtimeStatusService InstanceRuntimeStatusService) *SyncService { return &SyncService{ - instanceRepo: instanceRepo, - podService: k8s.NewPodService(), - interval: 5 * time.Second, // Sync every 5 seconds for more responsive status updates - stopChan: make(chan struct{}), + instanceRepo: instanceRepo, + runtimeStatusService: runtimeStatusService, + podService: k8s.NewPodService(), + interval: 5 * time.Second, // Sync every 5 seconds for more responsive status updates + stopChan: make(chan struct{}), } } @@ -115,6 +117,7 @@ func (s *SyncService) syncInstance(ctx context.Context, instance *models.Instanc if err := s.instanceRepo.Update(instance); err != nil { fmt.Printf("Error updating instance %d status: %v\n", instance.ID, err) } else { + s.updateInfraStatus(instance.ID, nextStatus) // Broadcast status update GetHub().BroadcastInstanceStatus(instance.UserID, instance) } @@ -133,6 +136,7 @@ func (s *SyncService) syncInstance(ctx context.Context, instance *models.Instanc instance.Status = desiredStatus needsUpdate = true } + s.updateInfraStatus(instance.ID, desiredStatus) // Update Pod IP if changed if pod.Status.PodIP != "" { @@ -167,6 +171,31 @@ func (s *SyncService) syncInstance(ctx context.Context, instance *models.Instanc } } +func (s *SyncService) updateInfraStatus(instanceID int, instanceStatus string) { + if s.runtimeStatusService == nil { + return + } + infraStatus := mapInstanceStatusToInfraStatus(instanceStatus) + if err := s.runtimeStatusService.UpsertInfraStatus(instanceID, infraStatus); err != nil { + fmt.Printf("Error updating runtime infra status for instance %d: %v\n", instanceID, err) + } +} + +func mapInstanceStatusToInfraStatus(instanceStatus string) string { + switch instanceStatus { + case "running": + return "ready" + case "stopped": + return "stopped" + case "error": + return "error" + case "creating": + return "creating" + default: + return "creating" + } +} + func mapPodToInstanceStatus(pod *corev1.Pod) string { if pod == nil { return "error" diff --git a/backend/internal/utils/response.go b/backend/internal/utils/response.go index ddd6caf..ccabf6b 100644 --- a/backend/internal/utils/response.go +++ b/backend/internal/utils/response.go @@ -58,16 +58,18 @@ func HandleError(c *gin.Context, err error) { Error(c, http.StatusConflict, errStr) case "display name already exists": Error(c, http.StatusConflict, errStr) - case "unsupported instance type", "image is required", "display name is required", "provider type is required", "base URL is required", "provider model name is required", "input price must be non-negative", "output price must be non-negative", "base URL is invalid", "automatic model discovery for azure-openai is not supported yet", "provider discovery is not supported", "model is required", "messages are required", "streaming is not supported yet", "provider type is not supported yet", "trace id is required", "event type is required", "message is required", "risk hit record is incomplete", "rule id is required", "rule display name is required", "rule pattern is required", "rule pattern is invalid", "risk severity is invalid", "risk action is invalid", "sample text is required", "secret ref format is invalid", "secret namespace is required in secret ref", "invalid openclaw resource type", "invalid openclaw config plan mode", "openclaw config resource name is required", "openclaw config resource key is invalid", "openclaw config schemaVersion is required", "openclaw config kind does not match resource type", "openclaw config format is required", "openclaw config content is required", "openclaw config content must be valid JSON", "openclaw config config payload is required", "openclaw config dependency type is invalid", "openclaw config dependency key is required", "openclaw config dependency is invalid", "openclaw config bundle name is required", "openclaw config bundle must include at least one resource", "openclaw config bundle resource id is required", "openclaw config bundle contains duplicate resources", "openclaw config bundle is required", "openclaw config bundle is disabled", "openclaw config bundle is empty", "at least one openclaw config resource must be selected", "openclaw config resource id is invalid", "openclaw config resource is disabled", "openclaw config bundle contains a disabled resource", "openclaw bootstrap payload is too large": + case "unsupported instance type", "image is required", "display name is required", "provider type is required", "base URL is required", "provider model name is required", "input price must be non-negative", "output price must be non-negative", "base URL is invalid", "automatic model discovery for azure-openai is not supported yet", "provider discovery is not supported", "model is required", "messages are required", "streaming is not supported yet", "provider type is not supported yet", "trace id is required", "event type is required", "message is required", "risk hit record is incomplete", "rule id is required", "rule display name is required", "rule pattern is required", "rule pattern is invalid", "risk severity is invalid", "risk action is invalid", "sample text is required", "secret ref format is invalid", "secret namespace is required in secret ref", "invalid openclaw resource type", "invalid openclaw config plan mode", "openclaw config resource name is required", "openclaw config resource key is invalid", "openclaw config schemaVersion is required", "openclaw config kind does not match resource type", "openclaw config format is required", "openclaw config content is required", "openclaw config content must be valid JSON", "openclaw config config payload is required", "openclaw config dependency type is invalid", "openclaw config dependency key is required", "openclaw config dependency is invalid", "openclaw config bundle name is required", "openclaw config bundle must include at least one resource", "openclaw config bundle resource id is required", "openclaw config bundle contains duplicate resources", "openclaw config bundle is required", "openclaw config bundle is disabled", "openclaw config bundle is empty", "at least one openclaw config resource must be selected", "openclaw config resource id is invalid", "openclaw config resource is disabled", "openclaw config bundle contains a disabled resource", "openclaw bootstrap payload is too large", "agent bootstrap token is required", "agent id is required", "unsupported agent protocol version", "invalid agent bootstrap token", "invalid instance command type", "invalid instance command finish status": Error(c, http.StatusBadRequest, errStr) - case "model is not active or does not exist", "openclaw config resource not found", "openclaw config bundle not found", "openclaw injection snapshot not found": + case "model is not active or does not exist", "openclaw config resource not found", "openclaw config bundle not found", "openclaw injection snapshot not found", "instance command not found", "instance config revision not found": Error(c, http.StatusNotFound, errStr) case "risk rule not found": Error(c, http.StatusNotFound, errStr) case "sensitive content requires an active secure model", "request was blocked by risk policy": Error(c, http.StatusForbidden, errStr) - case "invalid username or password", "account is disabled": + case "invalid username or password", "account is disabled", "invalid or expired agent session token": Error(c, http.StatusUnauthorized, errStr) + case "agent registration is only supported for openclaw instances", "agent id does not match session", "access denied": + Error(c, http.StatusForbidden, errStr) case "current password is incorrect": Error(c, http.StatusBadRequest, errStr) case "user not found", "model not found": diff --git a/frontend/src/components/InstanceAccess.tsx b/frontend/src/components/InstanceAccess.tsx index b916bf4..7704b93 100644 --- a/frontend/src/components/InstanceAccess.tsx +++ b/frontend/src/components/InstanceAccess.tsx @@ -1,6 +1,6 @@ -import { useState, useEffect, useCallback, useRef } from 'react'; -import { useI18n } from '../contexts/I18nContext'; -import { useInstanceDesktopAccess } from '../hooks/useInstanceDesktopAccess'; +import { memo, useState, useEffect, useCallback, useRef } from "react"; +import { useI18n } from "../contexts/I18nContext"; +import { useInstanceDesktopAccess } from "../hooks/useInstanceDesktopAccess"; interface InstanceAccessProps { instanceId: number; @@ -8,10 +8,48 @@ interface InstanceAccessProps { isRunning: boolean; } -export function InstanceAccess({ instanceId, instanceName, isRunning }: InstanceAccessProps) { +const desktopConnectPreferenceStore = new Map(); + +const DesktopIframeSurface = memo(function DesktopIframeSurface({ + frameHeightClass, + iframeRef, + embedUrl, + instanceName, + handleFrameLoad, + handleFrameError, +}: { + frameHeightClass: string; + iframeRef: React.RefObject; + embedUrl: string; + instanceName: string; + handleFrameLoad: (frame: HTMLIFrameElement | null) => void; + handleFrameError: () => void; +}) { + return ( +
+