diff --git a/apps/backend/internal/backendapp/services.go b/apps/backend/internal/backendapp/services.go index c113b3812e..711180dcfc 100644 --- a/apps/backend/internal/backendapp/services.go +++ b/apps/backend/internal/backendapp/services.go @@ -318,6 +318,12 @@ func (a *workflowStepGetterAdapter) GetNextStepByPosition(ctx context.Context, b return a.svc.GetNextStepByPosition(ctx, boardID, currentPosition) } +// ListStepsByWorkflow lets task admission find WIP-limited steps that pull +// newly-created work from a feeder step. +func (a *workflowStepGetterAdapter) ListStepsByWorkflow(ctx context.Context, workflowID string) ([]*wfmodels.WorkflowStep, error) { + return a.svc.ListStepsByWorkflow(ctx, workflowID) +} + // startStepResolverAdapter adapts workflow service to task service's StartStepResolver interface. type startStepResolverAdapter struct { svc *workflowservice.Service diff --git a/apps/backend/internal/task/service/service.go b/apps/backend/internal/task/service/service.go index ad813450d1..e415f4c471 100644 --- a/apps/backend/internal/task/service/service.go +++ b/apps/backend/internal/task/service/service.go @@ -139,6 +139,12 @@ type WorkflowStepGetter interface { GetNextStepByPosition(ctx context.Context, workflowID string, currentPosition int) (*wfmodels.WorkflowStep, error) } +// workflowStepLister is an optional extension used to find WIP steps that +// pull work from a feeder when new work arrives in that feeder. +type workflowStepLister interface { + ListStepsByWorkflow(ctx context.Context, workflowID string) ([]*wfmodels.WorkflowStep, error) +} + // PRTaskResolver resolves which tasks are associated with a GitHub PR number. // Implemented by the github service; injected so the task service can surface a // task by its PR number in search without coupling to the github schema. diff --git a/apps/backend/internal/task/service/service_tasks.go b/apps/backend/internal/task/service/service_tasks.go index 13d6277bb6..4da5f19f4d 100644 --- a/apps/backend/internal/task/service/service_tasks.go +++ b/apps/backend/internal/task/service/service_tasks.go @@ -167,11 +167,36 @@ func (s *Service) CreateTask(ctx context.Context, req *CreateTaskRequest) (*mode } s.publishTaskEvent(ctx, events.TaskCreated, task, nil) + s.pullTasksFromNewFeederWork(ctx, task.WorkflowID, task.WorkflowStepID) + if refreshed, err := s.tasks.GetTask(ctx, task.ID); err != nil { + s.logger.Warn("failed to refresh task after feeder pull", zap.String("task_id", task.ID), zap.Error(err)) + } else if refreshed != nil { + refreshed.Repositories = task.Repositories + task = refreshed + } s.logger.Info("task created", zap.String("task_id", task.ID), zap.String("title", task.Title)) return task, nil } +func (s *Service) pullTasksFromNewFeederWork(ctx context.Context, workflowID, feederStepID string) { + stepLister, ok := s.workflowStepGetter.(workflowStepLister) + if !ok || workflowID == "" || feederStepID == "" { + return + } + steps, err := stepLister.ListStepsByWorkflow(ctx, workflowID) + if err != nil { + s.logger.Warn("failed to list workflow steps after feeder task creation", + zap.String("workflow_id", workflowID), zap.Error(err)) + return + } + for _, step := range steps { + if step != nil && step.PullFromStepID == feederStepID { + s.pullNextTaskOnVacate(ctx, step.ID, "") + } + } +} + func (s *Service) createTaskWithCapacity(ctx context.Context, task *models.Task) error { if task.IsEphemeral || task.WorkflowStepID == "" { return s.tasks.CreateTask(ctx, task) diff --git a/apps/backend/internal/task/service/service_tasks_wip_test.go b/apps/backend/internal/task/service/service_tasks_wip_test.go index 30e8732df2..7d1c971534 100644 --- a/apps/backend/internal/task/service/service_tasks_wip_test.go +++ b/apps/backend/internal/task/service/service_tasks_wip_test.go @@ -112,6 +112,62 @@ func TestCreateTask_UnlimitedWIPStepPreservesCreation(t *testing.T) { } } +func TestCreateTask_PullsUnstartedFeederTaskIntoAvailableWIPStep(t *testing.T) { + svc, _, repo := createTestService(t) + ctx := context.Background() + seedWIPWorkflow(t, ctx, repo) + svc.SetWorkflowStepGetter(&fakeWorkflowStepGetter{steps: map[string]*wfmodels.WorkflowStep{ + "waiting-step": {ID: "waiting-step", WorkflowID: "wip-workflow", Name: "Waiting"}, + "review-step": {ID: "review-step", WorkflowID: "wip-workflow", Name: "Review", WIPLimit: 2, PullFromStepID: "waiting-step"}, + }}) + + created, err := svc.CreateTask(ctx, &CreateTaskRequest{ + WorkspaceID: "wip-workspace", WorkflowID: "wip-workflow", WorkflowStepID: "waiting-step", + Title: "Unstarted review task", + }) + if err != nil { + t.Fatalf("CreateTask: %v", err) + } + if created.WorkflowStepID != "review-step" { + t.Fatalf("returned workflow step = %q, want review-step", created.WorkflowStepID) + } + + stored, err := repo.GetTask(ctx, created.ID) + if err != nil { + t.Fatalf("GetTask: %v", err) + } + if stored.WorkflowStepID != "review-step" { + t.Fatalf("workflow step = %q, want review-step", stored.WorkflowStepID) + } +} + +func TestCreateTask_FeederTaskStaysWhenWIPStepFull(t *testing.T) { + svc, _, repo := createTestService(t) + ctx := context.Background() + seedWIPWorkflow(t, ctx, repo) + svc.SetWorkflowStepGetter(&fakeWorkflowStepGetter{steps: map[string]*wfmodels.WorkflowStep{ + "waiting-step": {ID: "waiting-step", WorkflowID: "wip-workflow", Name: "Waiting"}, + "review-step": {ID: "review-step", WorkflowID: "wip-workflow", Name: "Review", WIPLimit: 1, PullFromStepID: "waiting-step"}, + }}) + if err := repo.CreateTask(ctx, &models.Task{ + ID: "occupant", WorkspaceID: "wip-workspace", WorkflowID: "wip-workflow", + WorkflowStepID: "review-step", WIPAdmitted: true, State: v1.TaskStateCreated, + }); err != nil { + t.Fatalf("seed occupant: %v", err) + } + + created, err := svc.CreateTask(ctx, &CreateTaskRequest{ + WorkspaceID: "wip-workspace", WorkflowID: "wip-workflow", WorkflowStepID: "waiting-step", + Title: "Should stay in feeder", + }) + if err != nil { + t.Fatalf("CreateTask: %v", err) + } + if created.WorkflowStepID != "waiting-step" { + t.Fatalf("step = %q, want waiting-step (WIP full, must not promote)", created.WorkflowStepID) + } +} + func seedWIPWorkflow(t *testing.T, ctx context.Context, repo interface { CreateWorkspace(context.Context, *models.Workspace) error CreateWorkflow(context.Context, *models.Workflow) error diff --git a/apps/backend/internal/task/service/service_workflow.go b/apps/backend/internal/task/service/service_workflow.go index 6e1d650a5f..5e72690b92 100644 --- a/apps/backend/internal/task/service/service_workflow.go +++ b/apps/backend/internal/task/service/service_workflow.go @@ -677,7 +677,6 @@ func (s *Service) promoteFeederQueuedTask(ctx context.Context, candidate *models return s.promoteNextQueuedTask(ctx, targetStep, position, skipped) } s.publishTaskEvent(ctx, events.TaskUpdated, candidate, nil, oldWorkflowID) - s.publishTaskEvent(ctx, events.TaskQueuePromoted, candidate, nil) s.publishTaskMovedEvent(ctx, candidate, oldWorkflowID, fromStepID, targetStep.ID, "") return true } diff --git a/apps/backend/internal/task/service/service_workflow_test.go b/apps/backend/internal/task/service/service_workflow_test.go index e3d35f789a..b8a8547081 100644 --- a/apps/backend/internal/task/service/service_workflow_test.go +++ b/apps/backend/internal/task/service/service_workflow_test.go @@ -42,6 +42,16 @@ func (f *fakeWorkflowStepGetter) GetNextStepByPosition(_ context.Context, workfl return next, nil } +func (f *fakeWorkflowStepGetter) ListStepsByWorkflow(_ context.Context, workflowID string) ([]*wfmodels.WorkflowStep, error) { + steps := make([]*wfmodels.WorkflowStep, 0, len(f.steps)) + for _, step := range f.steps { + if step.WorkflowID == workflowID { + steps = append(steps, step) + } + } + return steps, nil +} + type testStepNotFound struct{} func (testStepNotFound) Error() string { return "step not found" } @@ -564,14 +574,21 @@ func TestService_MoveTaskPullsNextFeederTaskOnVacate(t *testing.T) { } movedEvents := 0 + queuePromotedEvents := 0 for _, event := range eventBus.GetPublishedEvents() { if event.Type == events.TaskMoved { movedEvents++ } + if event.Type == events.TaskQueuePromoted { + queuePromotedEvents++ + } } if movedEvents != 2 { t.Fatalf("task.moved events = %d, want 2", movedEvents) } + if queuePromotedEvents != 0 { + t.Fatalf("feeder promotion queue-promoted events = %d, want 0", queuePromotedEvents) + } } func TestService_MoveTaskPullSkipsBlockedFeederCandidate(t *testing.T) { diff --git a/apps/web/e2e/tests/workflow/workflow-steps.spec.ts b/apps/web/e2e/tests/workflow/workflow-steps.spec.ts index 6970368390..2378d7ebf0 100644 --- a/apps/web/e2e/tests/workflow/workflow-steps.spec.ts +++ b/apps/web/e2e/tests/workflow/workflow-steps.spec.ts @@ -60,4 +60,35 @@ test.describe("Workflow steps", () => { await expect(kanban.taskCardByTitle("Step A Task")).toBeVisible(); await expect(kanban.taskCardByTitle("Step B Task")).toBeVisible(); }); + + test("an unstarted feeder task fills available WIP capacity", async ({ + testPage, + apiClient, + seedData, + }) => { + const workflow = await apiClient.createWorkflow(seedData.workspaceId, "Feeder pull workflow"); + const waiting = await apiClient.createWorkflowStep(workflow.id, "Waiting", 0, { + is_start_step: true, + }); + const review = await apiClient.createWorkflowStep(workflow.id, "Review", 1); + await apiClient.updateWorkflowStep(review.id, { + wip_limit: 2, + pull_from_step_id: waiting.id, + }); + + await apiClient.createTask(seedData.workspaceId, "Unstarted feeder task", { + workflow_id: workflow.id, + workflow_step_id: waiting.id, + }); + await apiClient.saveUserSettings({ + workspace_id: seedData.workspaceId, + workflow_filter_id: workflow.id, + repository_ids: [], + }); + + const kanban = new KanbanPage(testPage); + await kanban.goto(); + await expect(kanban.taskCardInColumn("Unstarted feeder task", review.id)).toBeVisible(); + await expect(kanban.taskCardInColumn("Unstarted feeder task", waiting.id)).toHaveCount(0); + }); }); diff --git a/docs/public/tasks-and-workflows.md b/docs/public/tasks-and-workflows.md index 375ffa5731..81f32e30e2 100644 --- a/docs/public/tasks-and-workflows.md +++ b/docs/public/tasks-and-workflows.md @@ -180,7 +180,7 @@ New steps allow manual moves by default. **Show in command panel** also defaults | **Show in command panel** | Includes tasks in this step in the default, empty-search **Cmd+K** task list. Typed task search currently searches every step and can also return archived tasks, regardless of this setting. | | **Auto-archive** | Archives inactive tasks after the configured number of hours. Enabling it starts at 24 hours; the minimum is 1. | | **WIP limit** | Maximum admitted active, non-archived, non-ephemeral tasks in the step. `0` means unlimited. Overflow remains visible as queued cards; manual moves into a full step are still rejected. | -| **Pull from** | Optional one-hop feeder step. When capacity opens, Kandev promotes queued work from the destination first, then the feeder. A full feeder rejects new overflow creation. | +| **Pull from** | Optional one-hop feeder step. When capacity opens or eligible work arrives in the feeder, Kandev promotes queued work from the destination first, then the feeder. A full feeder rejects new overflow creation. | The WIP check also applies when a task is created. It runs for an explicit `workflow_step_id` and for the workflow's resolved start step, and the @@ -199,7 +199,7 @@ reservation and does not start an agent for them. Auto-archive is checked on a five-minute background interval and uses the task's last update time. Any task update postpones eligibility, so the archive is not guaranteed at the exact configured minute. Archiving, deleting, or moving an admitted task opens capacity and promotes the oldest queued card. Auto-archive affects the task itself, not its children. -Pull configuration rejects self-references, cycles, and cross-workflow feeders. Pulling runs when a task vacates the limited step and fills each newly open slot. Candidates are ordered by board position, then priority (`critical`, `high`, `medium`, `low`, `none`), queue time, creation time, and ID. A candidate whose move fails—for example because its session is running or starting—is skipped for that pull pass. +Pull configuration rejects self-references, cycles, and cross-workflow feeders. Pulling runs when a task vacates the limited step and when eligible work is created in its feeder, filling each available slot. Candidates are ordered by board position, then priority (`critical`, `high`, `medium`, `low`, `none`), queue time, creation time, and ID. A candidate whose move fails—for example because its session is running or starting—is skipped for that pull pass. ### Configure events and transitions