package api import ( "context" "encoding/json" "errors" "fmt" "net/http" "strings" "time" "github.com/google/uuid" "github.com/thepeterstone/claudomator/internal/event" "github.com/thepeterstone/claudomator/internal/storage" "github.com/thepeterstone/claudomator/internal/task" ) // Shared task-operation service layer. Both the REST handlers and the // chatbot-facing MCP tools call these methods so the two surfaces cannot drift. // Errors are typed so callers can map them to the right transport status. // errTaskNotFound is returned when a task ID does not resolve. REST maps it to // 404; MCP surfaces it as a tool error. var errTaskNotFound = errors.New("task not found") // conflictError signals an operation that is invalid for the task's current // state (REST 409). type conflictError struct{ msg string } func (e *conflictError) Error() string { return e.msg } func conflictf(format string, a ...any) error { return &conflictError{msg: fmt.Sprintf(format, a...)} } // badRequestError signals malformed or missing input (REST 400). type badRequestError struct{ msg string } func (e *badRequestError) Error() string { return e.msg } func badRequestf(format string, a ...any) error { return &badRequestError{msg: fmt.Sprintf(format, a...)} } // submitTaskSpec describes a task a chatbot wants created and immediately run. type submitTaskSpec struct { Name string Instructions string Project string RepositoryURL string Model string MaxBudgetUSD float64 Timeout time.Duration Tags []string } // submitTask creates a task and queues it for execution in one step. The // repository URL is resolved from the named project when not given explicitly, // because task.Validate requires it before the executor's lazy resolution runs. func (s *Server) submitTask(ctx context.Context, spec submitTaskSpec) (*task.Task, error) { if strings.TrimSpace(spec.Instructions) == "" { return nil, badRequestf("instructions are required") } repoURL := spec.RepositoryURL if repoURL == "" && spec.Project != "" { if proj, err := s.store.GetProject(spec.Project); err == nil { repoURL = proj.RemoteURL } } if repoURL == "" { return nil, badRequestf("repository_url is required (pass it directly or a project that resolves to one)") } name := strings.TrimSpace(spec.Name) if name == "" { name = firstLine(spec.Instructions, 72) } now := time.Now().UTC() t := &task.Task{ ID: uuid.New().String(), Name: name, Project: spec.Project, RepositoryURL: repoURL, Agent: task.AgentConfig{ Type: "claude", Model: spec.Model, Instructions: spec.Instructions, MaxBudgetUSD: spec.MaxBudgetUSD, }, Priority: task.PriorityNormal, Tags: spec.Tags, DependsOn: []string{}, Retry: task.RetryConfig{MaxAttempts: 1, Backoff: "exponential"}, State: task.StatePending, CreatedAt: now, UpdatedAt: now, } if spec.Timeout > 0 { t.Timeout.Duration = spec.Timeout } if t.Tags == nil { t.Tags = []string{} } if err := task.Validate(t); err != nil { return nil, badRequestf("%s", err.Error()) } if err := s.store.CreateTask(t); err != nil { return nil, err } if err := s.store.UpdateTaskStateBy(t.ID, task.StateQueued, event.ActorChatbot); err != nil { return nil, err } t.State = task.StateQueued // Use the server's long-lived lifecycle context, not the caller's request // context — the request (REST or chatbot MCP tool call) returns long // before the task finishes running, and its context is cancelled when it // does. Submitting with it would cancel the task's execution moments // after dispatch. Mirrors the pattern at server.go's other Submit call sites. if err := s.pool.Submit(s.ctx, t); err != nil { return nil, err } return t, nil } // acceptTask transitions a READY task to COMPLETED. func (s *Server) acceptTask(ctx context.Context, id string, actor event.Actor) error { t, err := s.store.GetTask(id) if err != nil { return errTaskNotFound } if !task.ValidTransition(t.State, task.StateCompleted) { return conflictf("task cannot be accepted from state %s", t.State) } if err := s.store.UpdateTaskStateBy(id, task.StateCompleted, actor); err != nil { return err } return nil } // rejectTask transitions a READY task back to PENDING with an optional comment. func (s *Server) rejectTask(id, comment string) error { t, err := s.store.GetTask(id) if err != nil { return errTaskNotFound } if !task.ValidTransition(t.State, task.StatePending) { return conflictf("task cannot be rejected from state %s", t.State) } return s.store.RejectTask(id, comment) } // cancelTask cancels a running/queued task, returning a human-readable result. func (s *Server) cancelTask(id string, actor event.Actor) (string, error) { tk, err := s.store.GetTask(id) if err != nil { return "", errTaskNotFound } if s.pool.Cancel(id) { return "task cancellation requested", nil } if !task.ValidTransition(tk.State, task.StateCancelled) { return "", conflictf("task cannot be cancelled from state %s", tk.State) } if err := s.store.UpdateTaskStateBy(id, task.StateCancelled, actor); err != nil { return "", err } return "task cancelled", nil } // answerTaskQuestion answers a BLOCKED task's clarification and resumes it. func (s *Server) answerTaskQuestion(ctx context.Context, id, answer string) error { tk, err := s.questionStore.GetTask(id) if err != nil { return errTaskNotFound } if tk.State != task.StateBlocked { return conflictf("task is not blocked") } if strings.TrimSpace(answer) == "" { return badRequestf("answer is required") } latest, err := s.questionStore.GetLatestExecution(id) if err != nil || latest.SessionID == "" { return fmt.Errorf("no resumable session found") } if tk.QuestionJSON != "" { var qData struct { Text string `json:"text"` Options []string `json:"options"` } if json.Unmarshal([]byte(tk.QuestionJSON), &qData) == nil { if appendErr := s.questionStore.AppendTaskInteraction(id, task.Interaction{ QuestionText: qData.Text, Options: qData.Options, Answer: answer, AskedAt: tk.UpdatedAt, }); appendErr != nil { s.logger.Error("failed to append interaction", "taskID", id, "error", appendErr) } } } if err := s.questionStore.UpdateTaskQuestion(id, ""); err != nil { return err } if err := s.questionStore.UpdateTaskState(id, task.StateQueued); err != nil { return err } resumeExec := &storage.Execution{ ID: uuid.New().String(), TaskID: id, ResumeSessionID: latest.SessionID, ResumeAnswer: answer, SandboxDir: latest.SandboxDir, } return s.pool.SubmitResume(ctx, tk, resumeExec) } // writeOpError maps a service-layer error to the appropriate REST status. func writeOpError(w http.ResponseWriter, err error) { switch { case errors.Is(err, errTaskNotFound): writeJSON(w, http.StatusNotFound, map[string]string{"error": "task not found"}) return case errors.Is(err, errStoryNotFound): writeJSON(w, http.StatusNotFound, map[string]string{"error": "story not found"}) return } var ce *conflictError if errors.As(err, &ce) { writeJSON(w, http.StatusConflict, map[string]string{"error": ce.Error()}) return } var be *badRequestError if errors.As(err, &be) { writeJSON(w, http.StatusBadRequest, map[string]string{"error": be.Error()}) return } writeJSON(w, http.StatusInternalServerError, map[string]string{"error": err.Error()}) } // firstLine returns the first non-empty line of s, truncated to max runes. func firstLine(s string, max int) string { line := s if i := strings.IndexByte(s, '\n'); i >= 0 { line = s[:i] } line = strings.TrimSpace(line) if r := []rune(line); len(r) > max { return strings.TrimSpace(string(r[:max])) } if line == "" { return "task" } return line }