summaryrefslogtreecommitdiff
path: root/internal/api/server.go
diff options
context:
space:
mode:
Diffstat (limited to 'internal/api/server.go')
-rw-r--r--internal/api/server.go141
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})
generated by cgit v1.2.3 (git 2.39.1) at 2026-07-16 08:02:29 +0000