dreampiler/kandev
1
0
Fork 0
Code Issues 43 Pull Requests Actions 2 Packages Projects Releases Wiki Activity

fix: pull unstarted feeder tasks into available WIP steps (#2003)

Browse Source
This commit is contained in:
Carlos Florêncio
2026-07-28 21:26:05 +01:00
committed by GitHub
parent 7598880e67
commit a65087e062
8 changed files with 143 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/backend/internal/backendapp/services.go
+6
View File
@@ -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
apps/backend/internal/task/service/service.go
+6
View File
@@ -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.
apps/backend/internal/task/service/service_tasks.go
+25
View File
@@ -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)
apps/backend/internal/task/service/service_tasks_wip_test.go
+56
View File
@@ -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
apps/backend/internal/task/service/service_workflow.go
-1
View File
@@ -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
}
apps/backend/internal/task/service/service_workflow_test.go
+17
View File
@@ -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) {
apps/web/e2e/tests/workflow/workflow-steps.spec.ts
+31
View File
@@ -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);
});
});
docs/public/tasks-and-workflows.md
+2 -2
View File
@@ -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