[CAP-9161] return clear error when locally running a nonexistent task (#320)

Fix error propagation when starting a task run for an invalid task identifier on the local dev server.

GitOrigin-RevId: 43a9c557a21d493e8b79bd565bc724d73198ab6d

Author: blue-keleher

pkg/workflows/apiserver/server.go

@@ -37,7 +37,7 @@ func handleError(w http.ResponseWriter, err error, statusCode int) {
 		log.Println("error marshalling error", err)
 		return
 	}
-	w.Write(errJSON)
+	_, _ = w.Write(errJSON)
 }
 
 func Start(handler *ServerHandler, port int) (*http.Server, error) {
@@ -153,7 +153,6 @@ func (h *ServerHandler) RunTask(w http.ResponseWriter, r *http.Request) {
 	if err != nil {
 		if _, ok := err.(*orchestrator.TaskNotFoundError); ok {
 			handleError(w, err, http.StatusNotFound)
-			w.Write([]byte(err.Error()))
 			return
 		}
 
@@ -175,16 +174,12 @@ func (h *ServerHandler) RunTask(w http.ResponseWriter, r *http.Request) {
 func (h *ServerHandler) ListTaskRuns(w http.ResponseWriter, r *http.Request) {
 	w.Header().Set("Content-Type", "application/json")
 
-	var params client.ListTaskRunsParams
-
-	params.TaskId = pointers.From([]string{r.URL.Query().Get("taskId")})
-	if params.TaskId == nil {
+	taskID := r.URL.Query().Get("taskId")
+	if taskID == "" {
 		handleError(w, fmt.Errorf("taskId is required"), http.StatusBadRequest)
-		w.Write([]byte("taskId is required"))
 		return
 	}
 
-	taskID := (*params.TaskId)[0]
 	json.NewEncoder(w).Encode(internal.ListTaskRuns(h.taskStore, taskID))
 }
 

pkg/workflows/orchestrator/coordinator.go

@@ -22,6 +22,7 @@ type StatusReporter interface {
 	TaskRunning(taskRun *store.TaskRun)
 	TaskCompleted(taskRun *store.TaskRun)
 	TaskFailed(taskRun *store.TaskRun)
+	TaskNotFound(taskIdentifier string)
 }
 
 type Coordinator struct {
@@ -114,6 +115,7 @@ func (c *Coordinator) StartTask(ctx context.Context, taskIdentifier string, inpu
 
 	task := c.store.GetTask(taskIdentifier)
 	if task == nil {
+		c.statusReporter.TaskNotFound(taskIdentifier)
 		return nil, &TaskNotFoundError{TaskIdentifier: taskIdentifier}
 	}
 

pkg/workflows/orchestrator/coordinator_test.go

@@ -39,11 +39,76 @@ func (noopStatusReporter) TaskEnqueued(taskRun *store.TaskRun)  {}
 func (noopStatusReporter) TaskRunning(taskRun *store.TaskRun)   {}
 func (noopStatusReporter) TaskCompleted(taskRun *store.TaskRun) {}
 func (noopStatusReporter) TaskFailed(taskRun *store.TaskRun)    {}
+func (noopStatusReporter) TaskNotFound(taskIdentifier string)   {}
 
 func (f *fakeServerFactory) NewHandler(socket net.Listener, input taskserver.GetInput200JSONResponse, getSubtaskResultFunc taskserver.GetSubtaskResultFunc, startSubtaskFunc taskserver.StartSubtaskFunc) *taskserver.ServerHandler {
 	return f.newHandler(socket, input, getSubtaskResultFunc, startSubtaskFunc)
 }
 
+type recordingStatusReporter struct {
+	noopStatusReporter
+	notFoundIdentifier string
+}
+
+func (r *recordingStatusReporter) TaskNotFound(taskIdentifier string) {
+	r.notFoundIdentifier = taskIdentifier
+}
+
+func TestStartTaskNotFound(t *testing.T) {
+	ctx := context.Background()
+
+	s := store.NewTaskStore()
+
+	tasks := []taskserver.Task{
+		{
+			Name: "test-task",
+		},
+	}
+
+	postTasksChan := make(chan taskserver.PostRegisterTasksRequestObject, 1)
+	postTasksChan <- taskserver.PostRegisterTasksRequestObject{
+		Body: &taskserver.PostRegisterTasksJSONRequestBody{
+			Tasks: tasks,
+		},
+	}
+
+	socketTracker, err := orchestrator.NewSocketTracker(ctx)
+	require.NoError(t, err)
+
+	reporter := &recordingStatusReporter{}
+	coordinator := orchestrator.NewCoordinator(
+		ctx,
+		s,
+		&fakeSdkExec{
+			startService: func(ctx context.Context, taskRunID string, socket string, mode orchestrator.Mode) (func() error, <-chan error, error) {
+				return func() error { return nil }, neverExits(), nil
+			},
+		},
+		socketTracker,
+		&fakeServerFactory{
+			newHandler: func(socket net.Listener, input taskserver.GetInput200JSONResponse, getSubtaskResultFunc taskserver.GetSubtaskResultFunc, startSubtaskFunc taskserver.StartSubtaskFunc) *taskserver.ServerHandler {
+				return &taskserver.ServerHandler{
+					Socket: socket,
+					Input:  input,
+					Channels: taskserver.ServerChannels{
+						PostCallback: make(chan taskserver.PostCallbackRequestObject),
+						PostTasks:    postTasksChan,
+					},
+				}
+			},
+		},
+		reporter,
+	)
+
+	_, err = coordinator.StartTask(ctx, "nonexistent-task", []byte{}, nil)
+	require.Error(t, err)
+
+	var taskNotFoundErr *orchestrator.TaskNotFoundError
+	require.ErrorAs(t, err, &taskNotFoundErr)
+	require.Equal(t, "nonexistent-task", taskNotFoundErr.TaskIdentifier)
+	require.Equal(t, "nonexistent-task", reporter.notFoundIdentifier)
+}
+
 func TestStartTask(t *testing.T) {
 	ctx := context.Background()
 

pkg/workflows/orchestrator/statusreporter.go

@@ -85,6 +85,11 @@ func (r *PrintStatusReporter) TaskFailed(taskRun *store.TaskRun) {
 	r.print(r.formatWithDuration("%s %s: %s%s", taskRun), r.taskLabel(taskRun), status, r.describeTaskRun(taskRun), details)
 }
 
+func (r *PrintStatusReporter) TaskNotFound(taskIdentifier string) {
+	status := renderstyle.Status.Foreground(renderstyle.ColorError).Render("Not found")
+	r.print("Task %s: %s", status, taskIdentifier)
+}
+
 func (r *PrintStatusReporter) print(format string, args ...any) {
 	if r.printFn == nil {
 		return