diff --git a/agent/cmd/claw-agent/main.go b/agent/cmd/claw-agent/main.go new file mode 100644 index 00000000..8bc5a948 --- /dev/null +++ b/agent/cmd/claw-agent/main.go @@ -0,0 +1,121 @@ +package main + +import ( + "context" + "flag" + "fmt" + "log" + "os" + "os/signal" + "syscall" + + "github.com/mojomast/uberclawcontrol/agent/internal/clawdeck" + "github.com/mojomast/uberclawcontrol/agent/internal/config" + "github.com/mojomast/uberclawcontrol/agent/internal/orchestrator" + "github.com/mojomast/uberclawcontrol/agent/internal/runner" +) + +var ( + version = "dev" +) + +func main() { + showVersion := flag.Bool("version", false, "show version") + flag.Parse() + + if *showVersion { + fmt.Printf("claw-agent %s\n", version) + os.Exit(0) + } + + log.Printf("claw-agent %s starting", version) + + cfg, err := config.Load() + if err != nil { + log.Fatalf("failed to load config: %v", err) + } + + client := clawdeck.NewClient(cfg.APIURL) + + agentID, token, err := cfg.LoadPersistedToken() + if err != nil { + log.Printf("warning: failed to load persisted token: %v", err) + } + + if token == "" && cfg.JoinToken == "" { + log.Fatal("no persisted token found and CLAWDECK_JOIN_TOKEN not set") + } + + if token != "" { + client.SetToken(token) + client.SetAgentID(agentID) + log.Printf("using persisted token for agent %d", agentID) + } else { + log.Printf("registering agent with join token") + resp, err := client.Register(cfg.JoinToken, clawdeck.AgentInfo{ + Name: cfg.AgentInfo.Name, + Hostname: cfg.AgentInfo.Hostname, + HostUID: cfg.AgentInfo.HostUID, + Platform: cfg.AgentInfo.Platform, + Version: version, + Tags: cfg.AgentInfo.Tags, + Metadata: cfg.AgentInfo.Metadata, + }) + if err != nil { + log.Fatalf("failed to register: %v", err) + } + + agentID = resp.Agent.ID + log.Printf("registered agent id=%d name=%s", resp.Agent.ID, resp.Agent.Name) + + if err := cfg.SaveToken(resp.Agent.ID, resp.AgentToken); err != nil { + log.Printf("warning: failed to persist token: %v", err) + } + } + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + sigChan := make(chan os.Signal, 1) + signal.Notify(sigChan, syscall.SIGINT, syscall.SIGTERM) + + go func() { + sig := <-sigChan + log.Printf("received signal %v, shutting down", sig) + cancel() + }() + + heartbeatRunner := runner.NewHeartbeatRunner(client, agentID, cfg.HeartbeatDelay) + taskRunner := runner.NewTaskRunner(client, cfg.TaskPollDelay, runner.NewStubExecutor()) + commandRunner := orchestrator.NewCommandRunner(client, cfg.CommandPollDelay) + + errChan := make(chan error, 3) + + go func() { + if err := heartbeatRunner.Run(ctx); err != nil && err != context.Canceled { + errChan <- fmt.Errorf("heartbeat: %w", err) + } + }() + + go func() { + if err := taskRunner.Run(ctx); err != nil && err != context.Canceled { + errChan <- fmt.Errorf("task: %w", err) + } + }() + + go func() { + if err := commandRunner.Run(ctx); err != nil && err != context.Canceled { + errChan <- fmt.Errorf("command: %w", err) + } + }() + + select { + case <-ctx.Done(): + log.Printf("shutting down") + case err := <-errChan: + log.Printf("runner error: %v", err) + cancel() + } + + log.Printf("claw-agent stopped") +} diff --git a/agent/go.mod b/agent/go.mod new file mode 100644 index 00000000..94338edf --- /dev/null +++ b/agent/go.mod @@ -0,0 +1,9 @@ +module github.com/mojomast/uberclawcontrol/agent + +go 1.22 + +require ( + golang.org/x/term v0.17.0 +) + +require golang.org/x/sys v0.17.0 // indirect diff --git a/agent/internal/clawdeck/client.go b/agent/internal/clawdeck/client.go new file mode 100644 index 00000000..b4b45b98 --- /dev/null +++ b/agent/internal/clawdeck/client.go @@ -0,0 +1,202 @@ +package clawdeck + +import ( + "bytes" + "encoding/json" + "fmt" + "io" + "net/http" + "time" +) + +type Client struct { + baseURL string + httpClient *http.Client + token string + agentID int64 +} + +func NewClient(baseURL string) *Client { + return &Client{ + baseURL: baseURL, + httpClient: &http.Client{ + Timeout: 30 * time.Second, + }, + } +} + +func (c *Client) SetToken(token string) { + c.token = token +} + +func (c *Client) SetAgentID(id int64) { + c.agentID = id +} + +func (c *Client) AgentID() int64 { + return c.agentID +} + +func (c *Client) Register(joinToken string, info AgentInfo) (*RegisterResponse, error) { + req := RegisterRequest{ + JoinToken: joinToken, + Agent: info, + } + + var resp RegisterResponse + if err := c.doRequest("POST", "/api/v1/agents/register", req, &resp, false); err != nil { + return nil, fmt.Errorf("register: %w", err) + } + + c.token = resp.AgentToken + c.agentID = resp.Agent.ID + + return &resp, nil +} + +func (c *Client) Heartbeat(agentID int64, status string, metadata map[string]any) (*HeartbeatResponse, error) { + req := HeartbeatRequest{ + Status: status, + Metadata: metadata, + } + + var resp HeartbeatResponse + path := fmt.Sprintf("/api/v1/agents/%d/heartbeat", agentID) + if err := c.doRequest("POST", path, req, &resp, true); err != nil { + return nil, fmt.Errorf("heartbeat: %w", err) + } + + return &resp, nil +} + +func (c *Client) GetNextTask() (*Task, error) { + var resp Task + if err := c.doRequest("GET", "/api/v1/tasks/next", nil, &resp, true); err != nil { + if isNoContent(err) { + return nil, nil + } + return nil, fmt.Errorf("get next task: %w", err) + } + return &resp, nil +} + +func (c *Client) ClaimTask(taskID int64) (*Task, error) { + var resp Task + path := fmt.Sprintf("/api/v1/tasks/%d/claim", taskID) + if err := c.doRequest("PATCH", path, nil, &resp, true); err != nil { + return nil, fmt.Errorf("claim task: %w", err) + } + return &resp, nil +} + +func (c *Client) UpdateTask(taskID int64, updates TaskUpdateRequest) (*Task, error) { + var resp Task + path := fmt.Sprintf("/api/v1/tasks/%d", taskID) + if err := c.doRequest("PATCH", path, updates, &resp, true); err != nil { + return nil, fmt.Errorf("update task: %w", err) + } + return &resp, nil +} + +func (c *Client) CompleteTask(taskID int64) (*Task, error) { + var resp Task + path := fmt.Sprintf("/api/v1/tasks/%d/complete", taskID) + if err := c.doRequest("PATCH", path, nil, &resp, true); err != nil { + return nil, fmt.Errorf("complete task: %w", err) + } + return &resp, nil +} + +func (c *Client) GetNextCommand() (*Command, error) { + var resp Command + if err := c.doRequest("GET", "/api/v1/agent_commands/next", nil, &resp, true); err != nil { + if isNoContent(err) { + return nil, nil + } + return nil, fmt.Errorf("get next command: %w", err) + } + return &resp, nil +} + +func (c *Client) AckCommand(commandID int64) (*Command, error) { + var resp Command + path := fmt.Sprintf("/api/v1/agent_commands/%d/ack", commandID) + if err := c.doRequest("PATCH", path, CommandAckRequest{}, &resp, true); err != nil { + return nil, fmt.Errorf("ack command: %w", err) + } + return &resp, nil +} + +func (c *Client) CompleteCommand(commandID int64, result map[string]any) (*Command, error) { + req := CommandCompleteRequest{Result: result} + var resp Command + path := fmt.Sprintf("/api/v1/agent_commands/%d/complete", commandID) + if err := c.doRequest("PATCH", path, req, &resp, true); err != nil { + return nil, fmt.Errorf("complete command: %w", err) + } + return &resp, nil +} + +type noContentError struct{} + +func (e *noContentError) Error() string { + return "no content" +} + +func isNoContent(err error) bool { + _, ok := err.(*noContentError) + return ok +} + +func (c *Client) doRequest(method, path string, body any, out any, requireAuth bool) error { + var reqBody io.Reader + if body != nil { + data, err := json.Marshal(body) + if err != nil { + return fmt.Errorf("marshaling request: %w", err) + } + reqBody = bytes.NewReader(data) + } + + url := c.baseURL + path + req, err := http.NewRequest(method, url, reqBody) + if err != nil { + return fmt.Errorf("creating request: %w", err) + } + + req.Header.Set("Content-Type", "application/json") + if requireAuth && c.token != "" { + req.Header.Set("Authorization", "Bearer "+c.token) + } + + resp, err := c.httpClient.Do(req) + if err != nil { + return fmt.Errorf("executing request: %w", err) + } + defer resp.Body.Close() + + if resp.StatusCode == http.StatusNoContent { + return &noContentError{} + } + + respBody, err := io.ReadAll(resp.Body) + if err != nil { + return fmt.Errorf("reading response: %w", err) + } + + if resp.StatusCode >= 400 { + var errResp ErrorResponse + if json.Unmarshal(respBody, &errResp) == nil && errResp.Error != "" { + return fmt.Errorf("api error (%d): %s", resp.StatusCode, errResp.Error) + } + return fmt.Errorf("api error (%d): %s", resp.StatusCode, string(respBody)) + } + + if out != nil && len(respBody) > 0 { + if err := json.Unmarshal(respBody, out); err != nil { + return fmt.Errorf("unmarshaling response: %w", err) + } + } + + return nil +} diff --git a/agent/internal/clawdeck/types.go b/agent/internal/clawdeck/types.go new file mode 100644 index 00000000..e037adb9 --- /dev/null +++ b/agent/internal/clawdeck/types.go @@ -0,0 +1,110 @@ +package clawdeck + +import "time" + +type Agent struct { + ID int64 `json:"id"` + UserID int64 `json:"user_id"` + Name string `json:"name"` + Status string `json:"status"` + Hostname string `json:"hostname"` + HostUID string `json:"host_uid"` + Platform string `json:"platform"` + Version string `json:"version"` + Tags []string `json:"tags"` + Metadata map[string]any `json:"metadata"` + LastHeartbeatAt *time.Time `json:"last_heartbeat_at"` + CreatedAt time.Time `json:"created_at"` + UpdatedAt time.Time `json:"updated_at"` +} + +type Task struct { + ID int64 `json:"id"` + Name string `json:"name"` + Description string `json:"description"` + Priority string `json:"priority"` + Status string `json:"status"` + Blocked bool `json:"blocked"` + Tags []string `json:"tags"` + Completed bool `json:"completed"` + CompletedAt *time.Time `json:"completed_at"` + DueDate *time.Time `json:"due_date"` + Position int `json:"position"` + AssignedToAgent bool `json:"assigned_to_agent"` + AssignedAt *time.Time `json:"assigned_at"` + AssignedAgentID *int64 `json:"assigned_agent_id"` + AgentClaimedAt *time.Time `json:"agent_claimed_at"` + ClaimedByAgentID *int64 `json:"claimed_by_agent_id"` + BoardID int64 `json:"board_id"` + URL string `json:"url"` + CreatedAt time.Time `json:"created_at"` + UpdatedAt time.Time `json:"updated_at"` +} + +type Command struct { + ID int64 `json:"id"` + AgentID int64 `json:"agent_id"` + Kind string `json:"kind"` + Payload map[string]any `json:"payload"` + State string `json:"state"` + Result map[string]any `json:"result"` + RequestedByUserID int64 `json:"requested_by_user_id"` + AckedAt *time.Time `json:"acked_at"` + CompletedAt *time.Time `json:"completed_at"` + CreatedAt time.Time `json:"created_at"` + UpdatedAt time.Time `json:"updated_at"` +} + +type RegisterRequest struct { + JoinToken string `json:"join_token"` + Agent AgentInfo `json:"agent"` +} + +type AgentInfo struct { + Name string `json:"name"` + Hostname string `json:"hostname"` + HostUID string `json:"host_uid"` + Platform string `json:"platform"` + Version string `json:"version"` + Tags []string `json:"tags"` + Metadata map[string]string `json:"metadata"` +} + +type RegisterResponse struct { + Agent Agent `json:"agent"` + AgentToken string `json:"agent_token"` +} + +type HeartbeatRequest struct { + Status string `json:"status"` + Version string `json:"version,omitempty"` + Platform string `json:"platform,omitempty"` + Metadata map[string]any `json:"metadata,omitempty"` +} + +type HeartbeatResponse struct { + Agent Agent `json:"agent"` + DesiredState DesiredState `json:"desired_state"` +} + +type DesiredState struct { + Action string `json:"action"` +} + +type TaskUpdateRequest struct { + Status *string `json:"status,omitempty"` + Description *string `json:"description,omitempty"` + Priority *string `json:"priority,omitempty"` + Blocked *bool `json:"blocked,omitempty"` + ActivityNote *string `json:"activity_note,omitempty"` +} + +type CommandAckRequest struct{} + +type CommandCompleteRequest struct { + Result map[string]any `json:"result"` +} + +type ErrorResponse struct { + Error string `json:"error"` +} diff --git a/agent/internal/config/config.go b/agent/internal/config/config.go new file mode 100644 index 00000000..c92c3b9b --- /dev/null +++ b/agent/internal/config/config.go @@ -0,0 +1,136 @@ +package config + +import ( + "encoding/json" + "fmt" + "os" + "path/filepath" + "time" +) + +const ( + DefaultAPIURL = "http://localhost:3000" + DefaultHeartbeatDelay = 30 * time.Second + DefaultTaskPollDelay = 5 * time.Second + DefaultCommandPollDelay = 5 * time.Second + TokenFilePermissions = 0600 +) + +type Config struct { + APIURL string + JoinToken string + AgentTokenPath string + HeartbeatDelay time.Duration + TaskPollDelay time.Duration + CommandPollDelay time.Duration + AgentInfo AgentInfo +} + +type AgentInfo struct { + Name string `json:"name"` + Hostname string `json:"hostname"` + HostUID string `json:"host_uid"` + Platform string `json:"platform"` + Version string `json:"version"` + Tags []string `json:"tags"` + Metadata map[string]string `json:"metadata"` +} + +type persistedToken struct { + AgentID int64 `json:"agent_id"` + Token string `json:"token"` + StoredAt string `json:"stored_at"` +} + +func Load() (*Config, error) { + cfg := &Config{ + APIURL: getEnv("CLAWDECK_API_URL", DefaultAPIURL), + JoinToken: os.Getenv("CLAWDECK_JOIN_TOKEN"), + AgentTokenPath: getEnv("CLAWDECK_AGENT_TOKEN_PATH", ""), + HeartbeatDelay: DefaultHeartbeatDelay, + TaskPollDelay: DefaultTaskPollDelay, + CommandPollDelay: DefaultCommandPollDelay, + AgentInfo: AgentInfo{ + Name: getEnv("CLAWDECK_AGENT_NAME", "claw-agent"), + Hostname: getEnv("CLAWDECK_HOSTNAME", ""), + HostUID: getEnv("CLAWDECK_HOST_UID", ""), + Platform: getEnv("CLAWDECK_PLATFORM", ""), + Version: getEnv("CLAWDECK_VERSION", "0.1.0"), + }, + } + + if cfg.AgentInfo.Hostname == "" { + if h, err := os.Hostname(); err == nil { + cfg.AgentInfo.Hostname = h + } + } + + return cfg, nil +} + +func (c *Config) LoadPersistedToken() (agentID int64, token string, err error) { + if c.AgentTokenPath == "" { + return 0, "", nil + } + + data, err := os.ReadFile(c.AgentTokenPath) + if err != nil { + if os.IsNotExist(err) { + return 0, "", nil + } + return 0, "", fmt.Errorf("reading token file: %w", err) + } + + var pt persistedToken + if err := json.Unmarshal(data, &pt); err != nil { + return 0, "", fmt.Errorf("parsing token file: %w", err) + } + + return pt.AgentID, pt.Token, nil +} + +func (c *Config) SaveToken(agentID int64, token string) error { + if c.AgentTokenPath == "" { + return nil + } + + pt := persistedToken{ + AgentID: agentID, + Token: token, + StoredAt: time.Now().UTC().Format(time.RFC3339), + } + + data, err := json.MarshalIndent(pt, "", " ") + if err != nil { + return fmt.Errorf("marshaling token: %w", err) + } + + dir := filepath.Dir(c.AgentTokenPath) + if err := os.MkdirAll(dir, 0700); err != nil { + return fmt.Errorf("creating token directory: %w", err) + } + + if err := os.WriteFile(c.AgentTokenPath, data, TokenFilePermissions); err != nil { + return fmt.Errorf("writing token file: %w", err) + } + + return nil +} + +func (c *Config) ClearToken() error { + if c.AgentTokenPath == "" { + return nil + } + + if err := os.Remove(c.AgentTokenPath); err != nil && !os.IsNotExist(err) { + return fmt.Errorf("removing token file: %w", err) + } + return nil +} + +func getEnv(key, fallback string) string { + if value := os.Getenv(key); value != "" { + return value + } + return fallback +} diff --git a/agent/internal/orchestrator/command_loop.go b/agent/internal/orchestrator/command_loop.go new file mode 100644 index 00000000..136f510e --- /dev/null +++ b/agent/internal/orchestrator/command_loop.go @@ -0,0 +1,96 @@ +package orchestrator + +import ( + "context" + "log" + "time" + + "github.com/mojomast/uberclawcontrol/agent/internal/clawdeck" +) + +type CommandRunner struct { + client *clawdeck.Client + interval time.Duration + handlers map[string]CommandHandler +} + +type CommandHandler func(ctx context.Context, cmd *clawdeck.Command) map[string]any + +func NewCommandRunner(client *clawdeck.Client, interval time.Duration) *CommandRunner { + if interval == 0 { + interval = 5 * time.Second + } + cr := &CommandRunner{ + client: client, + interval: interval, + handlers: make(map[string]CommandHandler), + } + cr.registerDefaultHandlers() + return cr +} + +func (c *CommandRunner) Run(ctx context.Context) error { + ticker := time.NewTicker(c.interval) + defer ticker.Stop() + + for { + select { + case <-ctx.Done(): + log.Printf("command runner stopping: %v", ctx.Err()) + return ctx.Err() + case <-ticker.C: + c.pollAndHandle(ctx) + } + } +} + +func (c *CommandRunner) pollAndHandle(ctx context.Context) { + cmd, err := c.client.GetNextCommand() + if err != nil { + log.Printf("failed to get next command: %v", err) + return + } + + if cmd == nil { + return + } + + log.Printf("received command id=%d kind=%s state=%s", cmd.ID, cmd.Kind, cmd.State) + + _, err = c.client.AckCommand(cmd.ID) + if err != nil { + log.Printf("failed to ack command %d: %v", cmd.ID, err) + return + } + log.Printf("acknowledged command id=%d", cmd.ID) + + result := c.dispatch(ctx, cmd) + + _, err = c.client.CompleteCommand(cmd.ID, result) + if err != nil { + log.Printf("failed to complete command %d: %v", cmd.ID, err) + return + } + log.Printf("completed command id=%d", cmd.ID) +} + +func (c *CommandRunner) dispatch(ctx context.Context, cmd *clawdeck.Command) map[string]any { + handler, ok := c.handlers[cmd.Kind] + if !ok { + log.Printf("no handler for command kind=%s", cmd.Kind) + return map[string]any{"success": false, "error": "unknown command kind"} + } + + return handler(ctx, cmd) +} + +func (c *CommandRunner) RegisterHandler(kind string, handler CommandHandler) { + c.handlers[kind] = handler +} + +func (c *CommandRunner) registerDefaultHandlers() { + c.RegisterHandler("drain", HandleDrain) + c.RegisterHandler("resume", HandleResume) + c.RegisterHandler("restart", HandleRestart) + c.RegisterHandler("upgrade", HandleUpgrade) +} diff --git a/agent/internal/orchestrator/handlers.go b/agent/internal/orchestrator/handlers.go new file mode 100644 index 00000000..42b1a884 --- /dev/null +++ b/agent/internal/orchestrator/handlers.go @@ -0,0 +1,48 @@ +package orchestrator + +import ( + "context" + "log" + + "github.com/mojomast/uberclawcontrol/agent/internal/clawdeck" +) + +func HandleDrain(ctx context.Context, cmd *clawdeck.Command) map[string]any { + log.Printf("handling drain command: %+v", cmd.Payload) + return map[string]any{ + "success": true, + "message": "drain initiated", + } +} + +func HandleResume(ctx context.Context, cmd *clawdeck.Command) map[string]any { + log.Printf("handling resume command: %+v", cmd.Payload) + return map[string]any{ + "success": true, + "message": "resumed accepting tasks", + } +} + +func HandleRestart(ctx context.Context, cmd *clawdeck.Command) map[string]any { + log.Printf("handling restart command: %+v", cmd.Payload) + return map[string]any{ + "success": false, + "error": "restart not implemented in stub handler", + } +} + +func HandleUpgrade(ctx context.Context, cmd *clawdeck.Command) map[string]any { + log.Printf("handling upgrade command: %+v", cmd.Payload) + version, _ := cmd.Payload["version"].(string) + if version == "" { + return map[string]any{ + "success": false, + "error": "version required", + } + } + return map[string]any{ + "success": false, + "error": "upgrade not implemented in stub handler", + "requested_version": version, + } +} diff --git a/agent/internal/runner/executor.go b/agent/internal/runner/executor.go new file mode 100644 index 00000000..1d3fd38f --- /dev/null +++ b/agent/internal/runner/executor.go @@ -0,0 +1,26 @@ +package runner + +import ( + "context" + + "github.com/mojomast/uberclawcontrol/agent/internal/clawdeck" +) + +type Executor interface { + Execute(ctx context.Context, task *clawdeck.Task) ExecutionResult +} + +type ExecutionResult struct { + Completed bool + Error error + Output string +} + +type StubExecutor struct{} + +func (s *StubExecutor) Execute(ctx context.Context, task *clawdeck.Task) ExecutionResult { + return ExecutionResult{ + Completed: true, + Output: "stub execution completed", + } +} diff --git a/agent/internal/runner/heartbeat.go b/agent/internal/runner/heartbeat.go new file mode 100644 index 00000000..a813774c --- /dev/null +++ b/agent/internal/runner/heartbeat.go @@ -0,0 +1,83 @@ +package runner + +import ( + "context" + "log" + "runtime" + "time" + + "github.com/mojomast/uberclawcontrol/agent/internal/clawdeck" +) + +type HeartbeatRunner struct { + client *clawdeck.Client + interval time.Duration + agentID int64 + lastStatus string +} + +func NewHeartbeatRunner(client *clawdeck.Client, agentID int64, interval time.Duration) *HeartbeatRunner { + if interval == 0 { + interval = 30 * time.Second + } + return &HeartbeatRunner{ + client: client, + interval: interval, + agentID: agentID, + } +} + +func (h *HeartbeatRunner) Run(ctx context.Context) error { + ticker := time.NewTicker(h.interval) + defer ticker.Stop() + + h.sendHeartbeat(ctx) + + for { + select { + case <-ctx.Done(): + log.Printf("heartbeat runner stopping: %v", ctx.Err()) + return ctx.Err() + case <-ticker.C: + h.sendHeartbeat(ctx) + } + } +} + +func (h *HeartbeatRunner) sendHeartbeat(ctx context.Context) { + metadata := h.collectMetadata() + status := "online" + if h.lastStatus != "" { + status = h.lastStatus + } + + resp, err := h.client.Heartbeat(h.agentID, status, metadata) + if err != nil { + log.Printf("heartbeat failed: %v", err) + return + } + + log.Printf("heartbeat ok: agent status=%s desired_state=%s", + resp.Agent.Status, resp.DesiredState.Action) + + if resp.DesiredState.Action != "" && resp.DesiredState.Action != "none" { + log.Printf("desired state action: %s", resp.DesiredState.Action) + } +} + +func (h *HeartbeatRunner) collectMetadata() map[string]any { + var memStats runtime.MemStats + runtime.ReadMemStats(&memStats) + + return map[string]any{ + "goroutines": runtime.NumGoroutine(), + "go_version": runtime.Version(), + "alloc_mb": memStats.Alloc / 1024 / 1024, + "sys_mb": memStats.Sys / 1024 / 1024, + "num_cpu": runtime.NumCPU(), + } +} + +func (h *HeartbeatRunner) SetStatus(status string) { + h.lastStatus = status +} diff --git a/agent/internal/runner/task_loop.go b/agent/internal/runner/task_loop.go new file mode 100644 index 00000000..b2f2668f --- /dev/null +++ b/agent/internal/runner/task_loop.go @@ -0,0 +1,86 @@ +package runner + +import ( + "context" + "log" + "time" + + "github.com/mojomast/uberclawcontrol/agent/internal/clawdeck" +) + +type TaskRunner struct { + client *clawdeck.Client + interval time.Duration + executor Executor +} + +func NewTaskRunner(client *clawdeck.Client, interval time.Duration, executor Executor) *TaskRunner { + if interval == 0 { + interval = 5 * time.Second + } + if executor == nil { + executor = &StubExecutor{} + } + return &TaskRunner{ + client: client, + interval: interval, + executor: executor, + } +} + +func (t *TaskRunner) Run(ctx context.Context) error { + ticker := time.NewTicker(t.interval) + defer ticker.Stop() + + for { + select { + case <-ctx.Done(): + log.Printf("task runner stopping: %v", ctx.Err()) + return ctx.Err() + case <-ticker.C: + t.pollAndExecute(ctx) + } + } +} + +func (t *TaskRunner) pollAndExecute(ctx context.Context) { + task, err := t.client.GetNextTask() + if err != nil { + log.Printf("failed to get next task: %v", err) + return + } + + if task == nil { + return + } + + log.Printf("received task id=%d name=%q status=%s", task.ID, task.Name, task.Status) + + if task.ClaimedByAgentID == nil { + claimed, err := t.client.ClaimTask(task.ID) + if err != nil { + log.Printf("failed to claim task %d: %v", task.ID, err) + return + } + task = claimed + log.Printf("claimed task id=%d", task.ID) + } + + result := t.executor.Execute(ctx, task) + + if result.Error != nil { + log.Printf("task %d execution failed: %v", task.ID, result.Error) + return + } + + if result.Completed { + _, err = t.client.CompleteTask(task.ID) + if err != nil { + log.Printf("failed to complete task %d: %v", task.ID, err) + return + } + log.Printf("task %d completed", task.ID) + } +} + + diff --git a/app/controllers/agent_commands_controller.rb b/app/controllers/agent_commands_controller.rb new file mode 100644 index 00000000..1d0ad6c3 --- /dev/null +++ b/app/controllers/agent_commands_controller.rb @@ -0,0 +1,25 @@ +class AgentCommandsController < ApplicationController + before_action :set_agent + + def create + @command = @agent.agent_commands.build(command_params) + @command.requested_by_user = current_user + @command.state = :pending + + if @command.save + redirect_to agent_path(@agent), notice: "#{@command.kind.humanize} command queued." + else + redirect_to agent_path(@agent), alert: "Failed to queue command: #{@command.errors.full_messages.join(', ')}" + end + end + + private + + def set_agent + @agent = current_user.agents.find(params[:agent_id]) + end + + def command_params + params.require(:agent_command).permit(:kind, :payload) + end +end diff --git a/app/controllers/agents_controller.rb b/app/controllers/agents_controller.rb new file mode 100644 index 00000000..a007c494 --- /dev/null +++ b/app/controllers/agents_controller.rb @@ -0,0 +1,19 @@ +class AgentsController < ApplicationController + before_action :set_agent, only: [:show] + + def index + @agents = current_user.agents.order(last_heartbeat_at: :desc) + end + + def show + @commands = @agent.agent_commands.order(created_at: :desc).limit(20) + @tasks_assigned = @agent.assigned_tasks.order(updated_at: :desc).limit(10) + @tasks_claimed = @agent.claimed_tasks.order(updated_at: :desc).limit(10) + end + + private + + def set_agent + @agent = current_user.agents.find(params[:id]) + end +end diff --git a/app/controllers/api/v1/agent_commands_controller.rb b/app/controllers/api/v1/agent_commands_controller.rb new file mode 100644 index 00000000..da0477f2 --- /dev/null +++ b/app/controllers/api/v1/agent_commands_controller.rb @@ -0,0 +1,108 @@ +module Api + module V1 + class AgentCommandsController < BaseController + before_action :set_agent, only: [ :enqueue ] + before_action :set_agent_command, only: [ :ack, :complete ] + before_action :require_current_agent!, only: [ :next, :ack, :complete ] + before_action :require_command_ownership!, only: [ :ack, :complete ] + + def enqueue + unless current_user.admin? || current_user.id == @agent.user_id + render json: { error: "Forbidden" }, status: :forbidden + return + end + + command = @agent.agent_commands.create!( + kind: params[:kind], + payload: params[:payload] || {}, + requested_by_user: current_user + ) + + render json: agent_command_json(command), status: :created + end + + def next + command = nil + + AgentCommand.transaction do + command = current_agent.agent_commands.pending + .order(created_at: :asc) + .lock("FOR UPDATE SKIP LOCKED") + .first + + if command + command.update!(state: :acknowledged, acked_at: Time.current) + end + end + + if command + render json: agent_command_json(command) + else + head :no_content + end + end + + def ack + unless @agent_command.pending? + render json: { error: "Command must be pending to acknowledge" }, status: :unprocessable_entity + return + end + + @agent_command.update!(state: :acknowledged, acked_at: Time.current) + render json: agent_command_json(@agent_command) + end + + def complete + unless @agent_command.acknowledged? + render json: { error: "Command must be acknowledged to complete" }, status: :unprocessable_entity + return + end + + @agent_command.update!( + state: :completed, + completed_at: Time.current, + result: params[:result] || {} + ) + render json: agent_command_json(@agent_command) + end + + private + + def set_agent + @agent = Agent.find(params[:id]) + end + + def set_agent_command + @agent_command = AgentCommand.find(params[:id]) + end + + def require_current_agent! + return if current_agent + + render json: { error: "Unauthorized" }, status: :unauthorized + end + + def require_command_ownership! + return if current_agent.id == @agent_command.agent_id + + render json: { error: "Forbidden" }, status: :forbidden + end + + def agent_command_json(command) + { + id: command.id, + agent_id: command.agent_id, + kind: command.kind, + payload: command.payload, + state: command.state, + result: command.result, + requested_by_user_id: command.requested_by_user_id, + acked_at: command.acked_at&.iso8601, + completed_at: command.completed_at&.iso8601, + created_at: command.created_at.iso8601, + updated_at: command.updated_at.iso8601 + } + end + end + end +end diff --git a/app/controllers/api/v1/agents_controller.rb b/app/controllers/api/v1/agents_controller.rb new file mode 100644 index 00000000..abc413f4 --- /dev/null +++ b/app/controllers/api/v1/agents_controller.rb @@ -0,0 +1,114 @@ +module Api + module V1 + class AgentsController < BaseController + skip_before_action :authenticate_api_token, only: :register + before_action :set_agent, only: [ :show, :update ] + before_action :set_agent_for_heartbeat, only: [ :heartbeat ] + before_action :require_current_agent!, only: :heartbeat + before_action :require_agent_self!, only: :heartbeat + + def register + join_token = JoinToken.consume!(register_join_token) + unless join_token + render json: { error: "Invalid join token" }, status: :unauthorized + return + end + + agent = join_token.user.agents.new(register_params) + if agent.save + _agent_token, plaintext_token = AgentToken.issue!(agent: agent, name: "Bootstrap") + render json: { agent: agent_json(agent), agent_token: plaintext_token }, status: :created + else + render json: { error: agent.errors.full_messages.join(", ") }, status: :unprocessable_entity + end + end + + def heartbeat + updates = { + last_heartbeat_at: Time.current, + status: params[:status].presence || :online + } + updates[:version] = params[:version] if params.key?(:version) + updates[:platform] = params[:platform] if params.key?(:platform) + updates[:metadata] = params[:metadata] if params.key?(:metadata) + + @agent.update!(updates) + render json: { + agent: agent_json(@agent), + desired_state: { action: "none" } + } + end + + def index + agents = current_user.agents.order(created_at: :desc) + render json: agents.map { |agent| agent_json(agent) } + end + + def show + render json: agent_json(@agent) + end + + def update + if @agent.update(update_params) + render json: agent_json(@agent) + else + render json: { error: @agent.errors.full_messages.join(", ") }, status: :unprocessable_entity + end + end + + private + + def set_agent + @agent = current_user.agents.find(params[:id]) + end + + def set_agent_for_heartbeat + @agent = Agent.find(params[:id]) + end + + def require_current_agent! + return if current_agent + + render json: { error: "Unauthorized" }, status: :unauthorized + end + + def require_agent_self! + return if current_agent.id == @agent.id + + render json: { error: "Forbidden" }, status: :forbidden + end + + def register_join_token + params[:join_token] || params.dig(:agent, :join_token) + end + + def register_params + params.fetch(:agent, ActionController::Parameters.new) + .permit(:name, :hostname, :host_uid, :platform, :version, tags: [], metadata: {}) + end + + def update_params + params.fetch(:agent, ActionController::Parameters.new) + .permit(:name, :status, tags: [], metadata: {}) + end + + def agent_json(agent) + { + id: agent.id, + user_id: agent.user_id, + name: agent.name, + status: agent.status, + hostname: agent.hostname, + host_uid: agent.host_uid, + platform: agent.platform, + version: agent.version, + tags: agent.tags || [], + metadata: agent.metadata || {}, + last_heartbeat_at: agent.last_heartbeat_at&.iso8601, + created_at: agent.created_at.iso8601, + updated_at: agent.updated_at.iso8601 + } + end + end + end +end diff --git a/app/controllers/api/v1/tasks_controller.rb b/app/controllers/api/v1/tasks_controller.rb index 83a717e3..05f5e26c 100644 --- a/app/controllers/api/v1/tasks_controller.rb +++ b/app/controllers/api/v1/tasks_controller.rb @@ -2,6 +2,7 @@ module Api module V1 class TasksController < BaseController before_action :set_task, only: [ :show, :update, :destroy, :complete, :claim, :unclaim, :assign, :unassign ] + before_action :require_current_agent!, only: [ :next, :claim, :unclaim ] # GET /api/v1/tasks/next - get next task for agent to work on # Returns highest priority unclaimed task in "up_next" status @@ -13,10 +14,29 @@ def next return end - @task = current_user.tasks - .where(status: :up_next, blocked: false, agent_claimed_at: nil) - .reorder(priority: :desc, position: :asc) - .first + if current_agent.draining? + head :no_content + return + end + + @task = nil + + Task.transaction do + @task = current_user.tasks + .eligible_for_agent(current_agent) + .reorder(priority: :desc, position: :asc) + .lock("FOR UPDATE SKIP LOCKED") + .first + + if @task + set_task_activity_info(@task) + @task.update!( + claimed_by_agent: current_agent, + agent_claimed_at: Time.current, + status: :in_progress + ) + end + end if @task render json: task_json(@task) @@ -44,14 +64,18 @@ def pending_attention # PATCH /api/v1/tasks/:id/claim - agent claims a task def claim set_task_activity_info(@task) - @task.update!(agent_claimed_at: Time.current, status: :in_progress) + @task.update!( + claimed_by_agent: current_agent, + agent_claimed_at: Time.current, + status: :in_progress + ) render json: task_json(@task) end # PATCH /api/v1/tasks/:id/unclaim - agent releases a task def unclaim set_task_activity_info(@task) - @task.update!(agent_claimed_at: nil) + @task.update!(claimed_by_agent: nil, agent_claimed_at: nil) render json: task_json(@task) end @@ -176,11 +200,19 @@ def set_task def set_task_activity_info(task) task.activity_source = "api" + task.actor_user = current_user + task.actor_agent = current_agent task.actor_name = request.headers["X-Agent-Name"] task.actor_emoji = request.headers["X-Agent-Emoji"] task.activity_note = params[:activity_note] || params.dig(:task, :activity_note) end + def require_current_agent! + return if current_agent + + render json: { error: "Unauthorized" }, status: :unauthorized + end + def task_params params.require(:task).permit(:name, :description, :priority, :due_date, :status, :blocked, :board_id, tags: []) end @@ -200,7 +232,9 @@ def task_json(task) position: task.position, assigned_to_agent: task.assigned_to_agent, assigned_at: task.assigned_at&.iso8601, + assigned_agent_id: task.assigned_agent_id, agent_claimed_at: task.agent_claimed_at&.iso8601, + claimed_by_agent_id: task.claimed_by_agent_id, board_id: task.board_id, url: "https://clawdeck.io/boards/#{task.board_id}/tasks/#{task.id}", created_at: task.created_at.iso8601, diff --git a/app/controllers/boards/tasks_controller.rb b/app/controllers/boards/tasks_controller.rb index 884d47d6..99d13481 100644 --- a/app/controllers/boards/tasks_controller.rb +++ b/app/controllers/boards/tasks_controller.rb @@ -59,7 +59,17 @@ def destroy def assign @task.activity_source = "web" - @task.assign_to_agent! + agent_id = params[:agent_id] + if agent_id.present? && agent_id != "" + @agent = current_user.agents.find_by(id: agent_id) + if @agent + @task.update!(assigned_agent_id: @agent.id, assigned_to_agent: true, assigned_at: Time.current) + else + @task.update!(assigned_agent_id: nil, assigned_to_agent: true, assigned_at: Time.current) + end + else + @task.update!(assigned_agent_id: nil, assigned_to_agent: true, assigned_at: Time.current) + end respond_to do |format| format.turbo_stream do render turbo_stream: [ @@ -73,7 +83,7 @@ def assign def unassign @task.activity_source = "web" - @task.unassign_from_agent! + @task.update!(assigned_agent_id: nil, assigned_to_agent: false, assigned_at: nil) respond_to do |format| format.turbo_stream do render turbo_stream: [ @@ -96,7 +106,7 @@ def set_task end def task_params - permitted = params.require(:task).permit(:name, :title, :description, :priority, :status, :blocked, :due_date, :completed, :agent_hint, tags: []) + permitted = params.require(:task).permit(:name, :title, :description, :priority, :status, :blocked, :due_date, :completed, :agent_hint, :assigned_agent_id, tags: []) # Allow 'title' as alias for 'name' permitted[:name] = permitted.delete(:title) if permitted[:title].present? && permitted[:name].blank? permitted diff --git a/app/controllers/concerns/api/token_authentication.rb b/app/controllers/concerns/api/token_authentication.rb index 1fe7d9de..0d062e8f 100644 --- a/app/controllers/concerns/api/token_authentication.rb +++ b/app/controllers/concerns/api/token_authentication.rb @@ -5,21 +5,29 @@ module TokenAuthentication included do before_action :authenticate_api_token after_action :track_api_usage - attr_reader :current_user + attr_reader :current_user, :current_agent end private def authenticate_api_token token = extract_token_from_header - @current_user = ApiToken.authenticate(token) + agent_token = AgentToken.authenticate(token) + + if agent_token + @current_agent = agent_token.agent + @current_user = @current_agent.user + else + @current_agent = nil + @current_user = ApiToken.authenticate(token) + end unless @current_user render json: { error: "Unauthorized" }, status: :unauthorized return end - update_agent_info_from_headers + update_agent_info_from_headers if @current_agent.nil? end def extract_token_from_header diff --git a/app/models/agent.rb b/app/models/agent.rb new file mode 100644 index 00000000..7fbb23a0 --- /dev/null +++ b/app/models/agent.rb @@ -0,0 +1,30 @@ +class Agent < ApplicationRecord + belongs_to :user + + has_many :assigned_tasks, + class_name: "Task", + foreign_key: :assigned_agent_id, + inverse_of: :assigned_agent, + dependent: :nullify + has_many :claimed_tasks, + class_name: "Task", + foreign_key: :claimed_by_agent_id, + inverse_of: :claimed_by_agent, + dependent: :nullify + has_many :agent_tokens, dependent: :destroy + has_many :agent_commands, dependent: :destroy + has_many :task_activities, + class_name: "TaskActivity", + foreign_key: :actor_agent_id, + inverse_of: :actor_agent, + dependent: :nullify + + enum :status, { + offline: 0, + online: 1, + draining: 2, + disabled: 3 + }, default: :offline + + validates :name, presence: true +end diff --git a/app/models/agent_command.rb b/app/models/agent_command.rb new file mode 100644 index 00000000..76b2d391 --- /dev/null +++ b/app/models/agent_command.rb @@ -0,0 +1,16 @@ +class AgentCommand < ApplicationRecord + belongs_to :agent + belongs_to :requested_by_user, class_name: "User", optional: true + + enum :state, { + pending: 0, + acknowledged: 1, + completed: 2, + failed: 3 + }, default: :pending + + validates :kind, presence: true + + scope :for_agent, ->(agent) { where(agent: agent) } + scope :pending_for, ->(agent) { for_agent(agent).pending } +end diff --git a/app/models/agent_token.rb b/app/models/agent_token.rb new file mode 100644 index 00000000..6bdbb3fd --- /dev/null +++ b/app/models/agent_token.rb @@ -0,0 +1,42 @@ +class AgentToken < ApplicationRecord + TOKEN_BYTES = 32 + + belongs_to :agent + + validates :token_digest, presence: true, uniqueness: true + + def self.issue!(agent:, name: nil) + plaintext_token = SecureRandom.hex(TOKEN_BYTES) + + agent_token = create!( + agent: agent, + name: name, + token_digest: digest_token(plaintext_token) + ) + + [ agent_token, plaintext_token ] + end + + def self.authenticate(plaintext_token) + return nil if plaintext_token.blank? + + candidate_digest = digest_token(plaintext_token) + agent_token = find_by(token_digest: candidate_digest) + return nil unless agent_token + return nil unless secure_digest_compare(agent_token.token_digest, candidate_digest) + + agent_token.touch(:last_used_at) + agent_token + end + + def self.digest_token(plaintext_token) + OpenSSL::Digest::SHA256.hexdigest(plaintext_token.to_s) + end + + def self.secure_digest_compare(stored_digest, candidate_digest) + return false if stored_digest.blank? || candidate_digest.blank? + return false unless stored_digest.bytesize == candidate_digest.bytesize + + ActiveSupport::SecurityUtils.secure_compare(stored_digest, candidate_digest) + end +end diff --git a/app/models/join_token.rb b/app/models/join_token.rb new file mode 100644 index 00000000..f91c568e --- /dev/null +++ b/app/models/join_token.rb @@ -0,0 +1,52 @@ +class JoinToken < ApplicationRecord + TOKEN_BYTES = 32 + + belongs_to :user + belongs_to :created_by_user, class_name: "User", optional: true, inverse_of: :created_join_tokens + + validates :token_digest, presence: true, uniqueness: true + validates :expires_at, presence: true + + def self.issue!(user:, created_by_user: nil, expires_in: 24.hours) + plaintext_token = SecureRandom.hex(TOKEN_BYTES) + + join_token = create!( + user: user, + created_by_user: created_by_user, + token_digest: digest_token(plaintext_token), + expires_at: Time.current + expires_in + ) + + [ join_token, plaintext_token ] + end + + def self.consume!(plaintext_token, user: nil) + return nil if plaintext_token.blank? + + candidate_digest = digest_token(plaintext_token) + join_token = find_by(token_digest: candidate_digest) + return nil unless join_token + return nil unless secure_digest_compare(join_token.token_digest, candidate_digest) + + join_token.with_lock do + return nil if user.present? && join_token.user_id != user.id + return nil if join_token.used_at.present? + return nil if join_token.expires_at <= Time.current + + join_token.update!(used_at: Time.current) + end + + join_token + end + + def self.digest_token(plaintext_token) + OpenSSL::Digest::SHA256.hexdigest(plaintext_token.to_s) + end + + def self.secure_digest_compare(stored_digest, candidate_digest) + return false if stored_digest.blank? || candidate_digest.blank? + return false unless stored_digest.bytesize == candidate_digest.bytesize + + ActiveSupport::SecurityUtils.secure_compare(stored_digest, candidate_digest) + end +end diff --git a/app/models/task.rb b/app/models/task.rb index 20556c6a..c5eb209d 100644 --- a/app/models/task.rb +++ b/app/models/task.rb @@ -1,6 +1,8 @@ class Task < ApplicationRecord belongs_to :user belongs_to :board + belongs_to :assigned_agent, class_name: "Agent", optional: true + belongs_to :claimed_by_agent, class_name: "Agent", optional: true has_many :activities, class_name: "TaskActivity", dependent: :destroy has_many :subtasks, dependent: :destroy @@ -12,7 +14,7 @@ class Task < ApplicationRecord validates :status, inclusion: { in: statuses.keys } # Activity tracking - must be declared before callbacks that use it - attr_accessor :activity_source, :actor_name, :actor_emoji, :activity_note + attr_accessor :activity_source, :actor_name, :actor_emoji, :activity_note, :actor_user, :actor_agent # Store activity_source before commit so it survives the transaction before_save :store_activity_source_for_broadcast @@ -35,6 +37,12 @@ class Task < ApplicationRecord scope :completed, -> { where(completed: true).reorder(completed_at: :desc) } scope :assigned_to_agent, -> { where(assigned_to_agent: true).reorder(assigned_at: :asc) } scope :unassigned, -> { where(assigned_to_agent: false) } + scope :eligible_for_agent, ->(agent) do + return none unless agent + + where(status: :up_next, blocked: false, claimed_by_agent_id: nil) + .where(assigned_agent_id: [ nil, agent.id ]) + end default_scope { order(completed: :asc, position: :asc) } # Agent assignment methods @@ -77,7 +85,15 @@ def track_completion_time end def record_creation_activity - TaskActivity.record_creation(self, source: activity_source || "web", actor_name: actor_name, actor_emoji: actor_emoji, note: activity_note) + TaskActivity.record_creation( + self, + source: activity_source || "web", + actor_user: actor_user || user, + actor_agent: actor_agent, + actor_name: actor_name, + actor_emoji: actor_emoji, + note: activity_note + ) end def record_update_activities @@ -86,12 +102,33 @@ def record_update_activities # Track status/column changes if saved_change_to_status? old_status, new_status = saved_change_to_status - TaskActivity.record_status_change(self, old_status: old_status, new_status: new_status, source: source, actor_name: actor_name, actor_emoji: actor_emoji, note: activity_note) + TaskActivity.record_status_change( + self, + old_status: old_status, + new_status: new_status, + source: source, + actor_user: actor_user || user, + actor_agent: actor_agent, + actor_name: actor_name, + actor_emoji: actor_emoji, + note: activity_note + ) end # Track field changes tracked_changes = saved_changes.slice(*TaskActivity::TRACKED_FIELDS) - TaskActivity.record_changes(self, tracked_changes, source: source, actor_name: actor_name, actor_emoji: actor_emoji, note: activity_note) if tracked_changes.any? + if tracked_changes.any? + TaskActivity.record_changes( + self, + tracked_changes, + source: source, + actor_user: actor_user || user, + actor_agent: actor_agent, + actor_name: actor_name, + actor_emoji: actor_emoji, + note: activity_note + ) + end end # Turbo Streams broadcasts for real-time updates diff --git a/app/models/task_activity.rb b/app/models/task_activity.rb index 3eb8e1f1..0fd6eb4a 100644 --- a/app/models/task_activity.rb +++ b/app/models/task_activity.rb @@ -1,18 +1,20 @@ class TaskActivity < ApplicationRecord belongs_to :task belongs_to :user, optional: true + belongs_to :actor_agent, class_name: "Agent", optional: true validates :action, presence: true ACTIONS = %w[created updated moved].freeze - TRACKED_FIELDS = %w[name due_date].freeze + TRACKED_FIELDS = %w[name due_date claimed_by_agent_id].freeze scope :recent, -> { order(created_at: :desc) } - def self.record_creation(task, source: "web", actor_name: nil, actor_emoji: nil, note: nil) + def self.record_creation(task, source: "web", actor_user: nil, actor_agent: nil, actor_name: nil, actor_emoji: nil, note: nil) create!( task: task, - user: task.user, + user: actor_user || task.user, + actor_agent: actor_agent, action: "created", source: source, actor_type: source == "api" ? "agent" : "user", @@ -22,10 +24,11 @@ def self.record_creation(task, source: "web", actor_name: nil, actor_emoji: nil, ) end - def self.record_status_change(task, old_status:, new_status:, source: "web", actor_name: nil, actor_emoji: nil, note: nil) + def self.record_status_change(task, old_status:, new_status:, source: "web", actor_user: nil, actor_agent: nil, actor_name: nil, actor_emoji: nil, note: nil) create!( task: task, - user: Current.user, + user: actor_user || Current.user || task.user, + actor_agent: actor_agent, action: "moved", field_name: "status", old_value: old_status, @@ -38,14 +41,15 @@ def self.record_status_change(task, old_status:, new_status:, source: "web", act ) end - def self.record_changes(task, changes, source: "web", actor_name: nil, actor_emoji: nil, note: nil) + def self.record_changes(task, changes, source: "web", actor_user: nil, actor_agent: nil, actor_name: nil, actor_emoji: nil, note: nil) TRACKED_FIELDS.each do |field| next unless changes.key?(field) old_val, new_val = changes[field] create!( task: task, - user: Current.user, + user: actor_user || Current.user || task.user, + actor_agent: actor_agent, action: "updated", field_name: field, old_value: format_value(field, old_val), diff --git a/app/models/user.rb b/app/models/user.rb index 51c0ab35..e348b3c8 100644 --- a/app/models/user.rb +++ b/app/models/user.rb @@ -4,7 +4,14 @@ class User < ApplicationRecord has_many :sessions, dependent: :destroy has_many :boards, dependent: :destroy has_many :tasks, dependent: :destroy + has_many :agents, dependent: :destroy has_many :api_tokens, dependent: :destroy + has_many :join_tokens, dependent: :destroy + has_many :created_join_tokens, + class_name: "JoinToken", + foreign_key: :created_by_user_id, + dependent: :nullify, + inverse_of: :created_by_user has_one_attached :avatar # Primary API token for agent integration diff --git a/app/views/agents/index.html.erb b/app/views/agents/index.html.erb new file mode 100644 index 00000000..a3e7e3bc --- /dev/null +++ b/app/views/agents/index.html.erb @@ -0,0 +1,72 @@ +<% content_for :title, "Agents - clawdeck" %> +<% @board_page = false %> + +
+
+
+

Agents

+

Manage your fleet of OpenClaw agents

+
+ <%= link_to settings_path(anchor: "agents"), class: "inline-flex items-center gap-2 px-3 py-2 rounded-lg text-xs font-semibold bg-white/[0.04] hover:bg-white/[0.08] text-[#888] transition-colors" do %> + + + + Register Agent + <% end %> +
+ + <% if @agents.any? %> +
+ <% @agents.each do |agent| %> + <%= link_to agent_path(agent), class: "block p-4 rounded-xl bg-white/[0.03] border border-white/[0.06] hover:bg-white/[0.05] hover:border-white/[0.1] transition-all" do %> +
+
+
+ <%= agent.status == "online" ? "๐ŸŸข" : agent.status == "draining" ? "๐ŸŸก" : agent.status == "disabled" ? "๐Ÿ”ด" : "โšซ" %> +
+
+
+ <%= agent.name %> + "><%= agent.status.humanize %> +
+
+ <% if agent.hostname.present? %> + <%= agent.hostname %> + <% end %> + <% if agent.version.present? %> + v<%= agent.version %> + <% end %> +
+
+
+
+ <% if agent.last_heartbeat_at %> +
+ <%= time_ago_in_words(agent.last_heartbeat_at) %> ago +
+ <% else %> +
Never connected
+ <% end %> + <% if agent.tags.any? %> +
+ <% agent.tags.first(3).each do |tag| %> + <%= tag %> + <% end %> +
+ <% end %> +
+
+ <% end %> + <% end %> +
+ <% else %> +
+
๐Ÿค–
+

No agents registered

+

Get started by registering an agent using a join token.

+ <%= link_to settings_path(anchor: "agents"), class: "inline-flex items-center gap-2 px-4 py-2 rounded-lg text-sm font-semibold bg-accent text-white hover:bg-accent/90 transition-colors" do %> + Register Agent + <% end %> +
+ <% end %> +
diff --git a/app/views/agents/show.html.erb b/app/views/agents/show.html.erb new file mode 100644 index 00000000..c2b69610 --- /dev/null +++ b/app/views/agents/show.html.erb @@ -0,0 +1,157 @@ +<% content_for :title, "#{@agent.name} - Agents - clawdeck" %> +<% @board_page = false %> + +
+ <%# Breadcrumb %> +
+ <%= link_to "Agents", agents_path, class: "hover:text-[#888] transition-colors" %> + + + + <%= @agent.name %> +
+ + <%# Header %> +
+
+
+ <%= @agent.status == "online" ? "๐ŸŸข" : @agent.status == "draining" ? "๐ŸŸก" : @agent.status == "disabled" ? "๐Ÿ”ด" : "โšซ" %> +
+
+
+

<%= @agent.name %>

+ "><%= @agent.status.humanize %> +
+
+ <% if @agent.hostname.present? %><%= @agent.hostname %><% end %> + <% if @agent.platform.present? %>ยท<%= @agent.platform %><% end %> + <% if @agent.version.present? %>ยทv<%= @agent.version %><% end %> +
+
+
+ + <%# Command buttons %> +
+ <% if @agent.online? || @agent.draining? %> + <% if @agent.online? %> + <%= button_to agent_commands_path(@agent), + params: { agent_command: { kind: "drain", payload: { reason: "manual" }.to_json } }, + class: "inline-flex items-center gap-1.5 px-3 py-2 rounded-lg text-xs font-semibold bg-yellow-500/20 text-yellow-400 hover:bg-yellow-500/30 transition-colors", + data: { confirm: "Drain this agent? It will stop accepting new tasks." } do %> + + + + Drain + <% end %> + <% end %> + <% if @agent.draining? %> + <%= button_to agent_commands_path(@agent), + params: { agent_command: { kind: "resume", payload: {}.to_json } }, + class: "inline-flex items-center gap-1.5 px-3 py-2 rounded-lg text-xs font-semibold bg-emerald-500/20 text-emerald-400 hover:bg-emerald-500/30 transition-colors", + data: { confirm: "Resume this agent? It will start accepting new tasks." } do %> + + + + Resume + <% end %> + <% end %> + <%= button_to agent_commands_path(@agent), + params: { agent_command: { kind: "restart", payload: {}.to_json } }, + class: "inline-flex items-center gap-1.5 px-3 py-2 rounded-lg text-xs font-semibold bg-white/[0.04] text-[#888] hover:bg-white/[0.08] transition-colors", + data: { confirm: "Restart OpenClaw on this agent?" } do %> + + + + Restart + <% end %> + <% end %> +
+
+ + <%# Info grid %> +
+ <%# Last heartbeat %> +
+
Last Heartbeat
+
+ <%= @agent.last_heartbeat_at ? time_ago_in_words(@agent.last_heartbeat_at) + " ago" : "Never" %> +
+
+ + <%# Tasks %> +
+
Active Tasks
+
+ <%= @tasks_claimed.count %> claimed + / <%= @tasks_assigned.count %> assigned +
+
+
+ + <%# Tags %> + <% if @agent.tags.any? %> +
+
Tags
+
+ <% @agent.tags.each do |tag| %> + <%= tag %> + <% end %> +
+
+ <% end %> + + <%# Metadata %> + <% if @agent.metadata.present? && @agent.metadata.any? %> +
+
Metadata
+
+ 
<%= JSON.pretty_generate(@agent.metadata) %>
+
+
+ <% end %> + + <%# Recent Commands %> +
+
Recent Commands
+ <% if @commands.any? %> +
+ <% @commands.each do |cmd| %> +
+
+ "><%= cmd.state.humanize %> + <%= cmd.kind.humanize %> +
+ <%= cmd.created_at.strftime("%b %d, %H:%M") %> +
+ <% end %> +
+ <% else %> +
No commands yet
+ <% end %> +
+ + <%# Tasks %> + <% if @tasks_claimed.any? || @tasks_assigned.any? %> +
+
Tasks
+
+ <% (@tasks_claimed + @tasks_assigned).uniq.first(10).each do |task| %> + <%= link_to board_path(task.board), class: "block p-3 rounded-lg bg-white/[0.03] border border-white/[0.06] hover:bg-white/[0.05] transition-colors" do %> +
+
+
<%= task.name %>
+
+ <%= task.board.name %> ยท <%= task.status.humanize %> + <% if task.claimed_by_agent_id == @agent.id %> + Claimed + <% end %> +
+
+ <%= task.status.humanize %> +
+ <% end %> + <% end %> +
+
+ <% end %> +
diff --git a/app/views/boards/_task_card.html.erb b/app/views/boards/_task_card.html.erb index f321468a..c009f4bd 100644 --- a/app/views/boards/_task_card.html.erb +++ b/app/views/boards/_task_card.html.erb @@ -20,27 +20,54 @@