path: root/internal/api/taskops.go

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
	if err := s.pool.Submit(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
	}
	if t.StoryID != "" {
		go s.pool.CheckStoryCompletion(ctx, t.StoryID)
	}
	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
	}
	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
}