summaryrefslogtreecommitdiff
path: root/internal/api/taskops.go
diff options
context:
space:
mode:
authorPeter Stone <thepeterstone@gmail.com>2026-06-03 23:01:39 +0000
committerPeter Stone <thepeterstone@gmail.com>2026-06-03 23:01:39 +0000
commit68cf1e24dd2ca2612b72babc4b35280cd3512f92 (patch)
tree14b73f21570a2f42ae729ca7e9676de6628d6819 /internal/api/taskops.go
parentb28cfc6ff288d083f6c8e9c055b69bfcadbceccc (diff)
parent7388337d3be5cc7f65ef547e30b2e39884dd165b (diff)
merge: integrate oss branch — budget gating, MCP back-channel, events, loopback-only bind
Key features from OSS branch: - Budget gating: rolling per-provider spend caps with BUDGET_EXCEEDED task state - Agent MCP back-channel: runners mint tokens; /mcp endpoint resolves them - Chatbot MCP server at /chatbot/mcp (requires api_token in config) - Event timeline: unified event stream replacing ad-hoc question/summary flows - Loopback-only default bind (127.0.0.1:8484); external_bind_allowed=true to expose - repository_url required on task creation (enforces traceability) - Auto-checker: spawns verification task after each execution Conflict resolutions: - serve.go: keep cfg.Runners.XEnabled() conditional registration from main + agentRegistry from oss - config.go: keep RunnersConfig from main + BudgetConfig/ExternalBindAllowed from oss - server.go: handleStaticFiles (base-path rewrite) kept; deduplicated duplicate from both sides - executor/claude.go: accepted oss deletion (ClaudeRunner replaced by ContainerRunner) - container_test.go: added noopChannel + AgentChannel parameter to Run call Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
Diffstat (limited to 'internal/api/taskops.go')
-rw-r--r--internal/api/taskops.go253
1 files changed, 253 insertions, 0 deletions
diff --git a/internal/api/taskops.go b/internal/api/taskops.go
new file mode 100644
index 0000000..881babe
--- /dev/null
+++ b/internal/api/taskops.go
@@ -0,0 +1,253 @@
+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
+}