diff options
| author | Peter Stone <thepeterstone@gmail.com> | 2026-06-03 23:01:39 +0000 |
|---|---|---|
| committer | Peter Stone <thepeterstone@gmail.com> | 2026-06-03 23:01:39 +0000 |
| commit | 68cf1e24dd2ca2612b72babc4b35280cd3512f92 (patch) | |
| tree | 14b73f21570a2f42ae729ca7e9676de6628d6819 /internal/api/server.go | |
| parent | b28cfc6ff288d083f6c8e9c055b69bfcadbceccc (diff) | |
| parent | 7388337d3be5cc7f65ef547e30b2e39884dd165b (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/server.go')
| -rw-r--r-- | internal/api/server.go | 141 |
1 files changed, 28 insertions, 113 deletions
diff --git a/internal/api/server.go b/internal/api/server.go index afb80ad..4af7325 100644 --- a/internal/api/server.go +++ b/internal/api/server.go @@ -13,6 +13,7 @@ import ( "time" "github.com/thepeterstone/claudomator/internal/config" + "github.com/thepeterstone/claudomator/internal/event" "github.com/thepeterstone/claudomator/internal/executor" "github.com/thepeterstone/claudomator/internal/llm" "github.com/thepeterstone/claudomator/internal/notify" @@ -40,6 +41,7 @@ type Server struct { taskLogStore taskLogStore // injectable for tests; defaults to store questionStore questionStore // injectable for tests; defaults to store pool *executor.Pool + registry *executor.Registry // per-task agent MCP token registry; mounts /mcp when set hub *Hub logger *slog.Logger mux *http.ServeMux @@ -62,7 +64,8 @@ type Server struct { pushStore pushSubscriptionStore dropsDir string llm *llm.Client - basePath string // URL prefix the UI is mounted at, e.g. "/claudomator" + budget budgetReporter // optional; per-provider spend headroom for GET /api/budget + basePath string // URL prefix the UI is mounted at, e.g. "/claudomator" } // SetAPIToken configures a bearer token that must be supplied to access the API. @@ -98,6 +101,7 @@ func (s *Server) SetBasePath(p string) { s.basePath = p } + // Pool returns the executor pool, for graceful shutdown by the caller. func (s *Server) Pool() *executor.Pool { return s.pool } @@ -109,7 +113,7 @@ func (s *Server) SetLLM(c *llm.Client) { } -func NewServer(store *storage.DB, pool *executor.Pool, logger *slog.Logger, claudeBinPath, geminiBinPath string) *Server { +func NewServer(store *storage.DB, pool *executor.Pool, registry *executor.Registry, logger *slog.Logger, claudeBinPath, geminiBinPath string) *Server { wd, _ := os.Getwd() s := &Server{ ctx: context.Background(), @@ -118,6 +122,7 @@ func NewServer(store *storage.DB, pool *executor.Pool, logger *slog.Logger, clau taskLogStore: store, questionStore: store, pool: pool, + registry: registry, hub: NewHub(), logger: logger, mux: http.NewServeMux(), @@ -140,7 +145,6 @@ func (s *Server) StartHub() { } func (s *Server) routes() { - s.mux.HandleFunc("POST /api/tasks/elaborate", s.handleElaborateTask) s.mux.HandleFunc("POST /api/tasks/validate", s.handleValidateTask) s.mux.HandleFunc("POST /api/tasks", s.handleCreateTask) s.mux.HandleFunc("GET /api/tasks", s.handleListTasks) @@ -152,8 +156,10 @@ func (s *Server) routes() { s.mux.HandleFunc("DELETE /api/tasks/{id}", s.handleDeleteTask) s.mux.HandleFunc("GET /api/tasks/{id}/subtasks", s.handleListSubtasks) s.mux.HandleFunc("GET /api/tasks/{id}/executions", s.handleListExecutions) + s.mux.HandleFunc("GET /api/tasks/{id}/events", s.handleListTaskEvents) s.mux.HandleFunc("GET /api/executions", s.handleListRecentExecutions) s.mux.HandleFunc("GET /api/stats", s.handleGetDashboardStats) + s.mux.HandleFunc("GET /api/budget", s.handleGetBudget) s.mux.HandleFunc("GET /api/agents/status", s.handleGetAgentStatus) s.mux.HandleFunc("GET /api/executions/{id}", s.handleGetExecution) s.mux.HandleFunc("GET /api/executions/{id}/log", s.handleGetExecutionLog) @@ -189,6 +195,16 @@ func (s *Server) routes() { s.mux.HandleFunc("GET /api/drops", s.handleListDrops) s.mux.HandleFunc("GET /api/drops/{filename}", s.handleGetDrop) s.mux.HandleFunc("POST /api/drops", s.handlePostDrop) + if s.registry != nil { + mcpHandler := executor.NewAgentMCPHandler(s.registry) + s.mux.Handle("POST /mcp", mcpHandler) + s.mux.Handle("GET /mcp", mcpHandler) + s.mux.Handle("DELETE /mcp", mcpHandler) + } + chatbotHandler := s.chatbotMCPHandler() + s.mux.Handle("POST /chatbot/mcp", chatbotHandler) + s.mux.Handle("GET /chatbot/mcp", chatbotHandler) + s.mux.Handle("DELETE /chatbot/mcp", chatbotHandler) s.mux.HandleFunc("GET /", s.handleStaticFiles) } @@ -280,41 +296,16 @@ func (s *Server) handleDeleteTask(w http.ResponseWriter, r *http.Request) { func (s *Server) handleCancelTask(w http.ResponseWriter, r *http.Request) { taskID := r.PathValue("id") - tk, err := s.store.GetTask(taskID) + msg, err := s.cancelTask(taskID, event.ActorUser) if err != nil { - writeJSON(w, http.StatusNotFound, map[string]string{"error": "task not found"}) - return - } - // If the task is actively running in the pool, cancel it there. - if s.pool.Cancel(taskID) { - writeJSON(w, http.StatusOK, map[string]string{"message": "task cancellation requested", "task_id": taskID}) - return - } - // For non-running tasks (PENDING, QUEUED), transition directly to CANCELLED. - if !task.ValidTransition(tk.State, task.StateCancelled) { - writeJSON(w, http.StatusConflict, map[string]string{"error": "task cannot be cancelled from state " + string(tk.State)}) - return - } - if err := s.store.UpdateTaskState(taskID, task.StateCancelled); err != nil { - writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "failed to cancel task"}) + writeOpError(w, err) return } - writeJSON(w, http.StatusOK, map[string]string{"message": "task cancelled", "task_id": taskID}) + writeJSON(w, http.StatusOK, map[string]string{"message": msg, "task_id": taskID}) } func (s *Server) handleAnswerQuestion(w http.ResponseWriter, r *http.Request) { taskID := r.PathValue("id") - - tk, err := s.questionStore.GetTask(taskID) - if err != nil { - writeJSON(w, http.StatusNotFound, map[string]string{"error": "task not found"}) - return - } - if tk.State != task.StateBlocked { - writeJSON(w, http.StatusConflict, map[string]string{"error": "task is not blocked"}) - return - } - var input struct { Answer string `json:"answer"` } @@ -322,61 +313,10 @@ func (s *Server) handleAnswerQuestion(w http.ResponseWriter, r *http.Request) { writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid JSON: " + err.Error()}) return } - if input.Answer == "" { - writeJSON(w, http.StatusBadRequest, map[string]string{"error": "answer is required"}) - return - } - - // Look up the session ID from the most recent execution. - latest, err := s.questionStore.GetLatestExecution(taskID) - if err != nil || latest.SessionID == "" { - writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "no resumable session found"}) - return - } - - // Record the Q&A interaction before clearing the question. - if tk.QuestionJSON != "" { - var qData struct { - Text string `json:"text"` - Options []string `json:"options"` - } - if jsonErr := json.Unmarshal([]byte(tk.QuestionJSON), &qData); jsonErr == nil { - interaction := task.Interaction{ - QuestionText: qData.Text, - Options: qData.Options, - Answer: input.Answer, - AskedAt: tk.UpdatedAt, - } - if appendErr := s.questionStore.AppendTaskInteraction(taskID, interaction); appendErr != nil { - s.logger.Error("failed to append interaction", "taskID", taskID, "error", appendErr) - } - } - } - - // Clear the question and transition to QUEUED. - if err := s.questionStore.UpdateTaskQuestion(taskID, ""); err != nil { - writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "failed to clear question"}) + if err := s.answerTaskQuestion(s.ctx, taskID, input.Answer); err != nil { + writeOpError(w, err) return } - if err := s.questionStore.UpdateTaskState(taskID, task.StateQueued); err != nil { - writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "failed to queue task"}) - return - } - - // Submit a resume execution. Carry the sandbox path so the runner uses - // the same working directory where Claude stored its session files. - resumeExec := &storage.Execution{ - ID: uuid.New().String(), - TaskID: taskID, - ResumeSessionID: latest.SessionID, - ResumeAnswer: input.Answer, - SandboxDir: latest.SandboxDir, - } - if err := s.pool.SubmitResume(s.ctx, tk, resumeExec); err != nil { - writeJSON(w, http.StatusServiceUnavailable, map[string]string{"error": err.Error()}) - return - } - writeJSON(w, http.StatusOK, map[string]string{"message": "task queued for resume", "task_id": taskID}) } @@ -702,40 +642,15 @@ func (s *Server) handleRunTask(w http.ResponseWriter, r *http.Request) { func (s *Server) handleAcceptTask(w http.ResponseWriter, r *http.Request) { id := r.PathValue("id") - t, err := s.store.GetTask(id) - if err != nil { - writeJSON(w, http.StatusNotFound, map[string]string{"error": "task not found"}) - return - } - if !task.ValidTransition(t.State, task.StateCompleted) { - writeJSON(w, http.StatusConflict, map[string]string{ - "error": fmt.Sprintf("task cannot be accepted from state %s", t.State), - }) - return - } - if err := s.store.UpdateTaskState(id, task.StateCompleted); err != nil { - writeJSON(w, http.StatusInternalServerError, map[string]string{"error": err.Error()}) + if err := s.acceptTask(r.Context(), id, event.ActorUser); err != nil { + writeOpError(w, err) return } - if t.StoryID != "" { - go s.pool.CheckStoryCompletion(r.Context(), t.StoryID) - } writeJSON(w, http.StatusOK, map[string]string{"message": "task accepted", "task_id": id}) } func (s *Server) handleRejectTask(w http.ResponseWriter, r *http.Request) { id := r.PathValue("id") - t, err := s.store.GetTask(id) - if err != nil { - writeJSON(w, http.StatusNotFound, map[string]string{"error": "task not found"}) - return - } - if !task.ValidTransition(t.State, task.StatePending) { - writeJSON(w, http.StatusConflict, map[string]string{ - "error": fmt.Sprintf("task cannot be rejected from state %s", t.State), - }) - return - } var input struct { Comment string `json:"comment"` } @@ -743,8 +658,8 @@ func (s *Server) handleRejectTask(w http.ResponseWriter, r *http.Request) { writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid JSON: " + err.Error()}) return } - if err := s.store.RejectTask(id, input.Comment); err != nil { - writeJSON(w, http.StatusInternalServerError, map[string]string{"error": err.Error()}) + if err := s.rejectTask(id, input.Comment); err != nil { + writeOpError(w, err) return } writeJSON(w, http.StatusOK, map[string]string{"message": "task rejected", "task_id": id}) |
