feat(coderd): add experimental tasks send endpoint (coder#19941)

Fixes coder/internal#902

Author: mafredri

coderd/aitasks.go

@@ -1,19 +1,25 @@
 package coderd
 
 import (
+	"bytes"
 	"context"
 	"database/sql"
+	"encoding/json"
 	"errors"
 	"fmt"
+	"io"
+	"net"
 	"net/http"
+	"net/url"
+	"path"
 	"slices"
 	"strings"
+	"time"
 
 	"github.com/go-chi/chi/v5"
 	"github.com/google/uuid"
 
 	"cdr.dev/slog"
-
 	"github.com/coder/coder/v2/coderd/audit"
 	"github.com/coder/coder/v2/coderd/database"
 	"github.com/coder/coder/v2/coderd/httpapi"
@@ -590,3 +596,288 @@ func (api *API) taskDelete(rw http.ResponseWriter, r *http.Request) {
 	// Delete build created successfully.
 	rw.WriteHeader(http.StatusAccepted)
 }
+
+// taskSend submits task input to the tasks sidebar app by dialing the agent
+// directly over the tailnet. We enforce ApplicationConnect RBAC on the
+// workspace and validate the sidebar app health.
+func (api *API) taskSend(rw http.ResponseWriter, r *http.Request) {
+	ctx := r.Context()
+
+	idStr := chi.URLParam(r, "id")
+	taskID, err := uuid.Parse(idStr)
+	if err != nil {
+		httpapi.Write(ctx, rw, http.StatusBadRequest, codersdk.Response{
+			Message: fmt.Sprintf("Invalid UUID %q for task ID.", idStr),
+		})
+		return
+	}
+
+	var req codersdk.TaskSendRequest
+	if !httpapi.Read(ctx, rw, r, &req) {
+		return
+	}
+	if req.Input == "" {
+		httpapi.Write(ctx, rw, http.StatusBadRequest, codersdk.Response{
+			Message: "Task input is required.",
+		})
+		return
+	}
+
+	if err = api.authAndDoWithTaskSidebarAppClient(r, taskID, func(ctx context.Context, client *http.Client, appURL *url.URL) error {
+		status, err := agentapiDoStatusRequest(ctx, client, appURL)
+		if err != nil {
+			return err
+		}
+
+		if status != "stable" {
+			return httperror.NewResponseError(http.StatusBadGateway, codersdk.Response{
+				Message: "Task app is not ready to accept input.",
+				Detail:  fmt.Sprintf("Status: %s", status),
+			})
+		}
+
+		var reqBody struct {
+			Content string `json:"content"`
+			Type    string `json:"type"`
+		}
+		reqBody.Content = req.Input
+		reqBody.Type = "user"
+
+		req, err := agentapiNewRequest(ctx, http.MethodPost, appURL, "message", reqBody)
+		if err != nil {
+			return err
+		}
+
+		resp, err := client.Do(req)
+		if err != nil {
+			return httperror.NewResponseError(http.StatusBadGateway, codersdk.Response{
+				Message: "Failed to reach task app endpoint.",
+				Detail:  err.Error(),
+			})
+		}
+		defer resp.Body.Close()
+
+		if resp.StatusCode != http.StatusOK {
+			body, _ := io.ReadAll(io.LimitReader(resp.Body, 128))
+			return httperror.NewResponseError(http.StatusBadGateway, codersdk.Response{
+				Message: "Task app rejected the message.",
+				Detail:  fmt.Sprintf("Upstream status: %d; Body: %s", resp.StatusCode, body),
+			})
+		}
+
+		// {"$schema":"http://localhost:3284/schemas/MessageResponseBody.json","ok":true}
+		// {"$schema":"http://localhost:3284/schemas/ErrorModel.json","title":"Unprocessable Entity","status":422,"detail":"validation failed","errors":[{"location":"body.type","value":"oof"}]}
+		var respBody map[string]any
+		if err := json.NewDecoder(resp.Body).Decode(&respBody); err != nil {
+			return httperror.NewResponseError(http.StatusBadGateway, codersdk.Response{
+				Message: "Failed to decode task app response body.",
+				Detail:  err.Error(),
+			})
+		}
+
+		if v, ok := respBody["status"].(string); !ok || v != "ok" {
+			return httperror.NewResponseError(http.StatusBadGateway, codersdk.Response{
+				Message: "Task app rejected the message.",
+				Detail:  fmt.Sprintf("Upstream response: %v", respBody),
+			})
+		}
+
+		return nil
+	}); err != nil {
+		httperror.WriteResponseError(ctx, rw, err)
+		return
+	}
+
+	rw.WriteHeader(http.StatusNoContent)
+}
+
+// authAndDoWithTaskSidebarAppClient centralizes the shared logic to:
+//
+//   - Fetch the task workspace
+//   - Authorize ApplicationConnect on the workspace
+//   - Validate the AI task and sidebar app health
+//   - Dial the agent and construct an HTTP client to the apps loopback URL
+//
+// The provided callback receives the context, an HTTP client that dials via the
+// agent, and the base app URL (as a value URL) to perform any request.
+func (api *API) authAndDoWithTaskSidebarAppClient(
+	r *http.Request,
+	taskID uuid.UUID,
+	do func(ctx context.Context, client *http.Client, appURL *url.URL) error,
+) error {
+	ctx := r.Context()
+
+	workspaceID := taskID
+	workspace, err := api.Database.GetWorkspaceByID(ctx, workspaceID)
+	if err != nil {
+		if httpapi.Is404Error(err) {
+			return httperror.ErrResourceNotFound
+		}
+		return httperror.NewResponseError(http.StatusInternalServerError, codersdk.Response{
+			Message: "Internal error fetching workspace.",
+			Detail:  err.Error(),
+		})
+	}
+
+	// Connecting to applications requires ApplicationConnect on the workspace.
+	if !api.Authorize(r, policy.ActionApplicationConnect, workspace) {
+		return httperror.ErrResourceNotFound
+	}
+
+	data, err := api.workspaceData(ctx, []database.Workspace{workspace})
+	if err != nil {
+		return httperror.NewResponseError(http.StatusInternalServerError, codersdk.Response{
+			Message: "Internal error fetching workspace resources.",
+			Detail:  err.Error(),
+		})
+	}
+	if len(data.builds) == 0 || len(data.templates) == 0 {
+		return httperror.ErrResourceNotFound
+	}
+	build := data.builds[0]
+	if build.HasAITask == nil || !*build.HasAITask || build.AITaskSidebarAppID == nil || *build.AITaskSidebarAppID == uuid.Nil {
+		return httperror.NewResponseError(http.StatusBadRequest, codersdk.Response{
+			Message: "Task is not configured with a sidebar app.",
+		})
+	}
+
+	// Find the sidebar app details to get the URL and validate app health.
+	sidebarAppID := *build.AITaskSidebarAppID
+	agentID, sidebarApp, ok := func() (uuid.UUID, codersdk.WorkspaceApp, bool) {
+		for _, res := range build.Resources {
+			for _, agent := range res.Agents {
+				for _, app := range agent.Apps {
+					if app.ID == sidebarAppID {
+						return agent.ID, app, true
+					}
+				}
+			}
+		}
+		return uuid.Nil, codersdk.WorkspaceApp{}, false
+	}()
+	if !ok {
+		return httperror.NewResponseError(http.StatusBadRequest, codersdk.Response{
+			Message: "Task sidebar app not found in latest build.",
+		})
+	}
+
+	// Return an informative error if the app isn't healthy rather than trying
+	// and failing.
+	switch sidebarApp.Health {
+	case codersdk.WorkspaceAppHealthDisabled:
+		// No health check, pass through.
+	case codersdk.WorkspaceAppHealthInitializing:
+		return httperror.NewResponseError(http.StatusServiceUnavailable, codersdk.Response{
+			Message: "Task sidebar app is initializing. Try again shortly.",
+		})
+	case codersdk.WorkspaceAppHealthUnhealthy:
+		return httperror.NewResponseError(http.StatusServiceUnavailable, codersdk.Response{
+			Message: "Task sidebar app is unhealthy.",
+		})
+	}
+
+	// Build the direct app URL and dial the agent.
+	if sidebarApp.URL == "" {
+		return httperror.NewResponseError(http.StatusInternalServerError, codersdk.Response{
+			Message: "Task sidebar app URL is not configured.",
+		})
+	}
+	parsedURL, err := url.Parse(sidebarApp.URL)
+	if err != nil {
+		return httperror.NewResponseError(http.StatusInternalServerError, codersdk.Response{
+			Message: "Internal error parsing task app URL.",
+			Detail:  err.Error(),
+		})
+	}
+	if parsedURL.Scheme != "http" {
+		return httperror.NewResponseError(http.StatusBadRequest, codersdk.Response{
+			Message: "Only http scheme is supported for direct agent-dial.",
+		})
+	}
+
+	dialCtx, dialCancel := context.WithTimeout(ctx, time.Second*30)
+	defer dialCancel()
+	agentConn, release, err := api.agentProvider.AgentConn(dialCtx, agentID)
+	if err != nil {
+		return httperror.NewResponseError(http.StatusBadGateway, codersdk.Response{
+			Message: "Failed to reach task app endpoint.",
+			Detail:  err.Error(),
+		})
+	}
+	defer release()
+
+	client := &http.Client{
+		Transport: &http.Transport{
+			DialContext: func(ctx context.Context, network, addr string) (net.Conn, error) {
+				return agentConn.DialContext(ctx, network, addr)
+			},
+		},
+	}
+	return do(ctx, client, parsedURL)
+}
+
+func agentapiNewRequest(ctx context.Context, method string, appURL *url.URL, appURLPath string, body any) (*http.Request, error) {
+	u := *appURL
+	u.Path = path.Join(appURL.Path, appURLPath)
+
+	var bodyReader io.Reader
+	if body != nil {
+		b, err := json.Marshal(body)
+		if err != nil {
+			return nil, httperror.NewResponseError(http.StatusBadRequest, codersdk.Response{
+				Message: "Failed to marshal task app request body.",
+				Detail:  err.Error(),
+			})
+		}
+		bodyReader = bytes.NewReader(b)
+	}
+
+	req, err := http.NewRequestWithContext(ctx, method, u.String(), bodyReader)
+	if err != nil {
+		return nil, httperror.NewResponseError(http.StatusBadRequest, codersdk.Response{
+			Message: "Failed to create task app request.",
+			Detail:  err.Error(),
+		})
+	}
+	req.Header.Set("Content-Type", "application/json")
+	req.Header.Set("Accept", "application/json")
+
+	return req, nil
+}
+
+func agentapiDoStatusRequest(ctx context.Context, client *http.Client, appURL *url.URL) (string, error) {
+	req, err := agentapiNewRequest(ctx, http.MethodGet, appURL, "status", nil)
+	if err != nil {
+		return "", err
+	}
+
+	resp, err := client.Do(req)
+	if err != nil {
+		return "", httperror.NewResponseError(http.StatusBadGateway, codersdk.Response{
+			Message: "Failed to reach task app endpoint.",
+			Detail:  err.Error(),
+		})
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode != http.StatusOK {
+		return "", httperror.NewResponseError(http.StatusBadGateway, codersdk.Response{
+			Message: "Task app status returned an error.",
+			Detail:  fmt.Sprintf("Status code: %d", resp.StatusCode),
+		})
+	}
+
+	// {"$schema":"http://localhost:3284/schemas/StatusResponseBody.json","status":"stable"}
+	var respBody struct {
+		Status string `json:"status"`
+	}
+
+	if err := json.NewDecoder(resp.Body).Decode(&respBody); err != nil {
+		return "", httperror.NewResponseError(http.StatusBadGateway, codersdk.Response{
+			Message: "Failed to decode task app status response body.",
+			Detail:  err.Error(),
+		})
+	}
+
+	return respBody.Status, nil
+}