fix: enforce workflow-step WIP limits on task creation (#1980)

This commit is contained in:
Carlos Florêncio
2026-07-27 16:14:13 +01:00
committed by GitHub
parent 9f09ef6f7c
commit 06d72a1215
30 changed files with 1239 additions and 87 deletions
+18 -7
View File
@@ -651,14 +651,12 @@ func (h *Handlers) handleCreateTask(ctx context.Context, msg *ws.Message) (*ws.M
})
if err != nil {
h.logger.Error("failed to create task", zap.Error(err))
// Defense-in-depth: resolveTaskRepositories already catches this for the
// MCP path, but non-MCP callers (UI, internal engine) reach here directly.
if errors.Is(err, service.ErrSubtaskDepthExceeded) ||
errors.Is(err, service.ErrInvalidTaskWorkflow) ||
isMCPWorkflowNotFoundError(err) {
return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, err.Error(), nil)
code := classifyCreateTaskError(err)
message := "Failed to create task"
if code != ws.ErrorCodeInternalError {
message = err.Error()
}
return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, "Failed to create task", nil)
return ws.NewError(msg.ID, msg.Action, code, message, nil)
}
if h.handoffSvc != nil && workspacePolicy.NeedsAttachment() {
@@ -681,6 +679,19 @@ func (h *Handlers) handleCreateTask(ctx context.Context, msg *ws.Message) (*ws.M
return ws.NewResponse(msg.ID, msg.Action, dto.FromTask(task))
}
func classifyCreateTaskError(err error) string {
switch {
case errors.Is(err, service.ErrWIPLimitExceeded):
return ws.ErrorCodeConflict
case errors.Is(err, service.ErrSubtaskDepthExceeded),
errors.Is(err, service.ErrInvalidTaskWorkflow),
isMCPWorkflowNotFoundError(err):
return ws.ErrorCodeValidation
default:
return ws.ErrorCodeInternalError
}
}
// taskRepoResult holds the output of resolveTaskRepositories.
type taskRepoResult struct {
Repos []service.TaskRepositoryInput
@@ -196,6 +196,13 @@ func TestClassifyAddBranchError_UnresolvedBaseBranchIsValidation(t *testing.T) {
}
}
func TestClassifyCreateTaskErrorMapsWIPLimitToConflict(t *testing.T) {
err := fmt.Errorf("create task failed: %w", workflowmodels.NewWIPLimitError("review", 2, 2))
if got := classifyCreateTaskError(err); got != ws.ErrorCodeConflict {
t.Fatalf("expected ErrorCodeConflict, got %q", got)
}
}
// TestHandleAddBranchToTask_RejectsMultipleLocators pins the mutual-exclusion
// check at the WS handler tier: supplying two of repository_id /
// repository_url / local_path is an agent mistake that previously got
@@ -2,6 +2,7 @@ package orchestrator
import (
"context"
"errors"
"fmt"
"strconv"
"strings"
@@ -19,6 +20,10 @@ import (
wfmodels "github.com/kandev/kandev/internal/workflow/models"
)
func isWatcherCapacityRejection(err error) bool {
return errors.Is(err, wfmodels.ErrWIPLimitExceeded)
}
const (
issueWatchIDKey = "issue_watch_id"
issueNumberKey = "issue_number"
@@ -329,10 +334,18 @@ func (s *Service) createReviewTask(ctx context.Context, evt *github.NewReviewPRE
repositories := s.resolveReviewRepository(ctx, evt.WorkspaceID, pr)
task, err := s.reviewTaskCreator.CreateReviewTask(ctx, buildReviewTaskRequest(evt, repositories, repoSlug))
if err != nil {
s.logger.Error("failed to create review task",
zap.String("review_watch_id", evt.ReviewWatchID),
zap.Int("pr_number", pr.Number),
zap.Error(err))
if isWatcherCapacityRejection(err) {
s.logger.Info("review workflow step at capacity; deferring review task",
zap.String("review_watch_id", evt.ReviewWatchID),
zap.Int("pr_number", pr.Number),
zap.String("workflow_step_id", evt.WorkflowStepID),
zap.Error(err))
} else {
s.logger.Error("failed to create review task",
zap.String("review_watch_id", evt.ReviewWatchID),
zap.Int("pr_number", pr.Number),
zap.Error(err))
}
s.releaseReviewPR(ctx, evt)
return
}
@@ -1343,10 +1356,18 @@ func (s *Service) createIssueTask(ctx context.Context, evt *github.NewIssueEvent
task, err := s.issueTaskCreator.CreateIssueTask(ctx, req)
if err != nil {
s.logger.Error("failed to create issue task",
zap.String(issueWatchIDKey, evt.IssueWatchID),
zap.Int(issueNumberKey, issue.Number),
zap.Error(err))
if isWatcherCapacityRejection(err) {
s.logger.Info("issue workflow step at capacity; deferring issue task",
zap.String(issueWatchIDKey, evt.IssueWatchID),
zap.Int(issueNumberKey, issue.Number),
zap.String("workflow_step_id", evt.WorkflowStepID),
zap.Error(err))
} else {
s.logger.Error("failed to create issue task",
zap.String(issueWatchIDKey, evt.IssueWatchID),
zap.Int(issueNumberKey, issue.Number),
zap.Error(err))
}
s.releaseIssueWatch(ctx, evt)
return
}
@@ -5,6 +5,7 @@ import (
"testing"
"github.com/kandev/kandev/internal/github"
"github.com/kandev/kandev/internal/orchestrator/watcher"
"github.com/kandev/kandev/internal/task/models"
wfmodels "github.com/kandev/kandev/internal/workflow/models"
)
@@ -13,11 +14,17 @@ import (
type countingReviewTaskCreator struct {
calls int
err error
errs []error
taskID string
}
func (c *countingReviewTaskCreator) CreateReviewTask(_ context.Context, _ *ReviewTaskRequest) (*models.Task, error) {
c.calls++
if len(c.errs) > 0 {
err := c.errs[0]
c.errs = c.errs[1:]
return nil, err
}
if c.err != nil {
return nil, c.err
}
@@ -137,3 +144,86 @@ func TestCreateReviewTask_ReleasesOnTaskCreationFailure(t *testing.T) {
t.Errorf("expected no Assign call after failure, got %d", ghSvc.assignCalls)
}
}
func TestGitHubWatcherTaskCreateError_ClassifiesWIPAsDeferred(t *testing.T) {
err := wfmodels.NewWIPLimitError("step1", 2, 2)
if !isWatcherCapacityRejection(err) {
t.Fatal("expected WIP-limit rejection to be classified as deferred capacity")
}
}
func TestCreateReviewTask_WIPRejectionIsRetriedOnLaterPoll(t *testing.T) {
svc, _ := setupReviewTaskTest(t)
ghSvc := &mockGitHubService{reserveReturn: true}
svc.SetGitHubService(ghSvc)
creator := &countingReviewTaskCreator{
errs: []error{wfmodels.NewWIPLimitError("step1", 2, 2)},
taskID: "task-after-capacity",
}
svc.SetReviewTaskCreator(creator)
svc.createReviewTask(context.Background(), newReviewEvent())
svc.createReviewTask(context.Background(), newReviewEvent())
if creator.calls != 2 {
t.Fatalf("expected one rejected attempt and one retry, got %d calls", creator.calls)
}
if ghSvc.releaseCalls != 1 {
t.Fatalf("expected only the rejected attempt to release its reservation, got %d", ghSvc.releaseCalls)
}
if ghSvc.assignCalls != 1 || ghSvc.assignedTaskID != "task-after-capacity" {
t.Fatalf("expected retry to assign task ID, calls=%d task=%q", ghSvc.assignCalls, ghSvc.assignedTaskID)
}
}
func TestReviewWatchLifecycle_BootReadyStaysOnReviewUntilTurnComplete(t *testing.T) {
ctx := context.Background()
repo := setupTestRepo(t)
seedSession(t, repo, "review-task", "review-session", "review")
steps := newMockStepGetter()
steps.steps["review"] = &wfmodels.WorkflowStep{
ID: "review", WorkflowID: "wf1", Name: "Review", Position: 0,
Events: wfmodels.StepEvents{
OnEnter: []wfmodels.OnEnterAction{{Type: wfmodels.OnEnterAutoStartAgent}},
OnTurnComplete: []wfmodels.OnTurnCompleteAction{{Type: wfmodels.OnTurnCompleteMoveToNext}},
},
}
steps.steps["done"] = &wfmodels.WorkflowStep{ID: "done", WorkflowID: "wf1", Name: "Done", Position: 1}
agentMgr := &mockAgentManager{repoForExecutionLookup: repo}
svc := createTestServiceWithAgent(repo, steps, newMockTaskRepo(), agentMgr)
if !svc.shouldAutoStartStep(ctx, "review") {
t.Fatal("review step should auto-start an admitted watcher task")
}
svc.handleAgentBootReady(ctx, watcher.AgentEventData{TaskID: "review-task", SessionID: "review-session"})
bootTask, err := repo.GetTask(ctx, "review-task")
if err != nil {
t.Fatalf("load task after boot-ready: %v", err)
}
if bootTask.WorkflowStepID != "review" {
t.Fatalf("boot-ready must not advance review task, got step %q", bootTask.WorkflowStepID)
}
if err := repo.UpdateTaskSessionState(ctx, "review-session", models.TaskSessionStateRunning, ""); err != nil {
t.Fatalf("mark review turn running: %v", err)
}
task, err := repo.GetTask(ctx, "review-task")
if err != nil {
t.Fatalf("reload task before turn completion: %v", err)
}
session, err := repo.GetTaskSession(ctx, "review-session")
if err != nil {
t.Fatalf("reload session before turn completion: %v", err)
}
if !svc.processOnTurnComplete(ctx, task, session) {
t.Fatal("first genuine turn completion should advance the review task")
}
completedTask, err := repo.GetTask(ctx, "review-task")
if err != nil {
t.Fatalf("load task after turn completion: %v", err)
}
if completedTask.WorkflowStepID != "done" {
t.Fatalf("expected turn completion to move task to done, got %q", completedTask.WorkflowStepID)
}
}
@@ -255,30 +255,12 @@ func (s *Service) executeStepTransition(ctx context.Context, taskID, sessionID s
s.setSessionWaitingForInput(ctx, taskID, sessionID)
return
}
if err := s.validateTransitionWIPLimit(ctx, task, targetStep); err != nil {
s.logger.Warn("workflow transition rejected by WIP limit",
zap.String("task_id", taskID),
zap.String("to_step_id", toStepID),
zap.Error(err))
s.setSessionWaitingForInput(ctx, taskID, sessionID)
return
}
// Process on_exit actions for the step we're leaving (before the step change).
// Freshly load the session since the caller may not have it (legacy path).
exitSession, exitErr := s.repo.GetTaskSession(ctx, sessionID)
if exitErr != nil {
s.logger.Warn("failed to load session for on_exit",
zap.String("session_id", sessionID), zap.Error(exitErr))
} else {
s.processOnExit(ctx, taskID, exitSession, fromStep)
}
// Update the task's workflow step
// Atomically admit the target step before exit side effects. A capacity
// rejection must leave both the task and its source-step session state intact.
task.WorkflowStepID = toStepID
task.UpdatedAt = time.Now().UTC()
if err := s.repo.UpdateTask(ctx, task); err != nil {
s.logger.Error("failed to move task to next workflow step",
if err := s.updateTransitionTaskWithCapacity(ctx, task, targetStep); err != nil {
s.logger.Warn("workflow transition rejected or failed",
zap.String("task_id", taskID),
zap.String("from_step", fromStep.Name),
zap.String("to_step", targetStep.Name),
@@ -287,6 +269,16 @@ func (s *Service) executeStepTransition(ctx context.Context, taskID, sessionID s
return
}
// Process on_exit only after the transition is durably admitted. Freshly
// load the session since the caller may not have it (legacy path).
exitSession, exitErr := s.repo.GetTaskSession(ctx, sessionID)
if exitErr != nil {
s.logger.Warn("failed to load session for on_exit",
zap.String("session_id", sessionID), zap.Error(exitErr))
} else {
s.processOnExit(ctx, taskID, exitSession, fromStep)
}
// Publish task updated event via the task service so the payload carries
// the full context (session counts, primary session, repositories).
s.publishTaskUpdated(ctx, task)
@@ -346,22 +338,25 @@ func (s *Service) executeStepTransition(ctx context.Context, taskID, sessionID s
}
}
func (s *Service) validateTransitionWIPLimit(ctx context.Context, task *models.Task, targetStep *wfmodels.WorkflowStep) error {
if targetStep == nil || targetStep.WIPLimit <= 0 || task.WorkflowStepID == targetStep.ID {
return nil
func (s *Service) updateTransitionTaskWithCapacity(
ctx context.Context,
task *models.Task,
targetStep *wfmodels.WorkflowStep,
) error {
if targetStep == nil || targetStep.WIPLimit <= 0 {
return s.repo.UpdateTask(ctx, task)
}
limitsRepo, ok := s.repo.(workflowMoveLimitsRepository)
limitedRepo, ok := s.repo.(workflowLimitedMoveRepository)
if !ok {
return fmt.Errorf("WIP limit cannot be checked for workflow step %s", targetStep.ID)
}
occupants, err := limitsRepo.CountTasksByWorkflowStepExcludingTask(ctx, targetStep.ID, task.ID)
if err != nil {
return fmt.Errorf("count target workflow step tasks: %w", err)
}
if occupants >= targetStep.WIPLimit {
return fmt.Errorf("WIP limit exceeded for workflow step %s: limit %d already occupied", targetStep.ID, targetStep.WIPLimit)
}
return nil
return limitedRepo.UpdateTaskIfWorkflowStepHasCapacity(
ctx,
task,
targetStep.ID,
task.ID,
targetStep.WIPLimit,
)
}
// handleTaskMoved handles manual task step changes (drag-and-drop, stepper "Move here").
@@ -2,6 +2,7 @@ package orchestrator
import (
"context"
"errors"
"sync"
"testing"
"time"
@@ -9,6 +10,7 @@ import (
"github.com/kandev/kandev/internal/events"
"github.com/kandev/kandev/internal/events/bus"
"github.com/kandev/kandev/internal/task/models"
wfmodels "github.com/kandev/kandev/internal/workflow/models"
)
// mockEventBus captures published events for assertion.
@@ -17,6 +19,92 @@ type mockEventBus struct {
events []publishedEvent
}
func TestUpdateTransitionTaskWithCapacity_RejectsFullLimitedStep(t *testing.T) {
ctx := context.Background()
repo := setupTestRepo(t)
seedSession(t, repo, "t1", "s1", "step1")
if err := repo.CreateTask(ctx, &models.Task{
ID: "occupant",
WorkspaceID: "ws1",
WorkflowID: "wf1",
WorkflowStepID: "step2",
Title: "Occupant",
State: "TODO",
Priority: "medium",
}); err != nil {
t.Fatalf("create occupant: %v", err)
}
svc := createTestService(repo, newMockStepGetter(), newMockTaskRepo())
task, err := repo.GetTask(ctx, "t1")
if err != nil {
t.Fatalf("load task: %v", err)
}
task.WorkflowStepID = "step2"
target := &wfmodels.WorkflowStep{ID: "step2", WorkflowID: "wf1", WIPLimit: 1}
err = svc.updateTransitionTaskWithCapacity(ctx, task, target)
if !errors.Is(err, wfmodels.ErrWIPLimitExceeded) {
t.Fatalf("error=%v, want typed WIP rejection", err)
}
stored, err := repo.GetTask(ctx, "t1")
if err != nil {
t.Fatalf("reload task: %v", err)
}
if stored.WorkflowStepID != "step1" {
t.Fatalf("task moved despite full target, got %q", stored.WorkflowStepID)
}
}
func TestExecuteStepTransition_FullTargetLeavesOnExitStateIntact(t *testing.T) {
ctx := context.Background()
repo := setupTestRepo(t)
seedSession(t, repo, "t1", "s1", "step1")
if err := repo.SetSessionMetadataKey(ctx, "s1", "plan_mode", true); err != nil {
t.Fatalf("enable plan mode: %v", err)
}
if err := repo.CreateTask(ctx, &models.Task{
ID: "occupant",
WorkspaceID: "ws1",
WorkflowID: "wf1",
WorkflowStepID: "step2",
Title: "Occupant",
State: "TODO",
Priority: "medium",
}); err != nil {
t.Fatalf("create occupant: %v", err)
}
steps := newMockStepGetter()
fromStep := &wfmodels.WorkflowStep{
ID: "step1", WorkflowID: "wf1", Name: "Source",
Events: wfmodels.StepEvents{OnExit: []wfmodels.OnExitAction{{
Type: wfmodels.OnExitDisablePlanMode,
}}},
}
steps.steps["step2"] = &wfmodels.WorkflowStep{
ID: "step2", WorkflowID: "wf1", Name: "Limited", WIPLimit: 1,
}
svc := createTestService(repo, steps, newMockTaskRepo())
svc.executeStepTransition(ctx, "t1", "s1", fromStep, "step2", false)
task, err := repo.GetTask(ctx, "t1")
if err != nil {
t.Fatalf("load task: %v", err)
}
if task.WorkflowStepID != "step1" {
t.Fatalf("task moved despite full target, got %q", task.WorkflowStepID)
}
session, err := repo.GetTaskSession(ctx, "s1")
if err != nil {
t.Fatalf("load session: %v", err)
}
if enabled, _ := session.Metadata["plan_mode"].(bool); !enabled {
t.Fatal("capacity rejection must not run source step's on_exit actions")
}
}
type publishedEvent struct {
Subject string
Event *bus.Event
@@ -2,11 +2,13 @@ package orchestrator
import (
"context"
"errors"
"sync"
"go.uber.org/zap"
"github.com/kandev/kandev/internal/common/logger"
workflowmodels "github.com/kandev/kandev/internal/workflow/models"
)
// WatcherDispatchCoordinator owns the cross-integration pipeline that turns
@@ -357,8 +359,13 @@ func (c *WatcherDispatchCoordinator) Dispatch(ctx context.Context, src WatcherSo
task, err := c.getTaskCreator().CreateIssueTask(ctx, req)
if err != nil {
c.logger.Error("watcher dispatch: create issue task failed",
zap.String("source", src.Name()), zap.Error(err))
if errors.Is(err, workflowmodels.ErrWIPLimitExceeded) {
c.logger.Info("watcher dispatch: workflow step at capacity; deferring issue task",
zap.String("source", src.Name()), zap.String("workflow_step_id", req.WorkflowStepID), zap.Error(err))
} else {
c.logger.Error("watcher dispatch: create issue task failed",
zap.String("source", src.Name()), zap.Error(err))
}
src.Release(ctx, evt)
return
}
@@ -9,6 +9,7 @@ import (
"github.com/kandev/kandev/internal/common/logger"
"github.com/kandev/kandev/internal/task/models"
workflowmodels "github.com/kandev/kandev/internal/workflow/models"
)
// nopLogger returns a *logger.Logger backed by zap.NewNop() for use in tests.
@@ -247,6 +248,26 @@ func TestCoordinator_Dispatch_CreateError_Releases(t *testing.T) {
}
}
func TestCoordinator_Dispatch_WIPLimitRejection_ReleasesWithoutErrorSemantics(t *testing.T) {
src := &fakeWatcherSource{
name: "linear",
reserveOK: true,
buildReq: &IssueTaskRequest{WorkspaceID: "ws-1", WorkflowStepID: "step-1"},
}
tc := &fakeTaskCreator{createErr: workflowmodels.NewWIPLimitError("step-1", 2, 2)}
starter := &fakeTaskStarter{}
c := newTestCoordinator(t, tc, true, starter)
c.Dispatch(context.Background(), src, "evt")
if src.recordedRelease != 1 {
t.Fatalf("expected release after WIP rejection, got %d", src.recordedRelease)
}
if src.recordedAttach != 0 || starter.called != 0 {
t.Fatal("expected WIP rejection to stop before attach and auto-start")
}
}
func TestCoordinator_Dispatch_NoAutoStart(t *testing.T) {
src := &fakeWatcherSource{
name: "linear",
@@ -127,10 +127,6 @@ func (s *workflowStore) ApplyTransition(ctx context.Context, taskID, sessionID,
if err != nil {
return fmt.Errorf("load target step for transition: %w", err)
}
if err := s.validateWIPLimit(ctx, task, targetStep); err != nil {
return err
}
// Keep WorkflowID in sync with the target step's owning workflow. Most
// callers transition within the same workflow (targetStep.WorkflowID ==
// task.WorkflowID already), but applyPendingMove uses this path for
@@ -180,24 +176,6 @@ func (s *workflowStore) updateTransitionTask(ctx context.Context, task *models.T
return limitedRepo.UpdateTaskIfWorkflowStepHasCapacity(ctx, task, targetStep.ID, task.ID, targetStep.WIPLimit)
}
func (s *workflowStore) validateWIPLimit(ctx context.Context, task *models.Task, targetStep *wfmodels.WorkflowStep) error {
if targetStep == nil || targetStep.WIPLimit <= 0 || task.WorkflowStepID == targetStep.ID {
return nil
}
limitsRepo, ok := s.repo.(workflowMoveLimitsRepository)
if !ok {
return fmt.Errorf("WIP limit cannot be checked for workflow step %s", targetStep.ID)
}
occupants, err := limitsRepo.CountTasksByWorkflowStepExcludingTask(ctx, targetStep.ID, task.ID)
if err != nil {
return fmt.Errorf("count target workflow step tasks: %w", err)
}
if occupants >= targetStep.WIPLimit {
return fmt.Errorf("WIP limit exceeded for workflow step %s: limit %d already occupied", targetStep.ID, targetStep.WIPLimit)
}
return nil
}
func (s *workflowStore) pullNextTaskOnVacate(ctx context.Context, vacatedStepID, excludeTaskID string) {
vacatedStep := s.pullEnabledStep(ctx, vacatedStepID)
if vacatedStep == nil {
@@ -17,6 +17,10 @@ func handleNotFound(c *gin.Context, log *logger.Logger, err error, fallback stri
c.JSON(http.StatusNotFound, gin.H{"error": fallback})
return
}
if errors.Is(err, service.ErrWIPLimitExceeded) {
c.JSON(http.StatusConflict, gin.H{"error": err.Error()})
return
}
if isValidationError(err) {
c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
return
@@ -5,11 +5,15 @@ import (
"errors"
"fmt"
"net"
"net/http"
"net/http/httptest"
"testing"
"github.com/gin-gonic/gin"
"github.com/kandev/kandev/internal/agent/runtime/lifecycle"
taskrepo "github.com/kandev/kandev/internal/task/repository/sqlite"
"github.com/kandev/kandev/internal/task/service"
wfmodels "github.com/kandev/kandev/internal/workflow/models"
)
// TestErrorsAreClassifiable verifies that the package's error-classification
@@ -42,6 +46,16 @@ func TestErrorsAreClassifiable(t *testing.T) {
t.Errorf("untyped lookalike must no longer classify")
}
})
t.Run("WIP limit is a conflict for task creation", func(t *testing.T) {
gin.SetMode(gin.TestMode)
rec := httptest.NewRecorder()
ctx, _ := gin.CreateTestContext(rec)
handleNotFound(ctx, newTestLogger(t), fmt.Errorf("create task: %w", wfmodels.NewWIPLimitError("review", 2, 2)), "task not created")
if rec.Code != http.StatusConflict {
t.Fatalf("status=%d, want %d", rec.Code, http.StatusConflict)
}
})
}
// TestIsTimeoutError pins the UX-classification contract for the prompt
@@ -75,6 +89,7 @@ func TestIsTimeoutError(t *testing.T) {
}
})
}
}
type timeoutNetErr struct{}
@@ -320,6 +320,14 @@ type missingWorkflowCreateTaskRepo struct {
captureCreateTaskRepo
}
type wipCreateTaskRepo struct {
captureCreateTaskRepo
}
func (*wipCreateTaskRepo) CreateTask(_ context.Context, _ *models.Task) error {
return wfmodels.NewWIPLimitError("review", 2, 2)
}
func (*missingWorkflowCreateTaskRepo) GetWorkflow(_ context.Context, id string) (*models.Workflow, error) {
return nil, fmt.Errorf("workflow not found: %s", id)
}
@@ -703,6 +711,34 @@ func TestWSCreateTaskReturnsValidationErrorForMissingWorkflow(t *testing.T) {
assert.Contains(t, payload.Message, "workflow not found")
}
func TestWSCreateTaskReturnsConflictForWIPLimit(t *testing.T) {
log := newTestLogger(t)
repo := &wipCreateTaskRepo{}
svc := service.NewService(service.Repos{
Workspaces: repo, Tasks: repo, TaskRepos: repo,
Workflows: repo, Messages: repo, Turns: repo,
Sessions: repo, GitSnapshots: repo, RepoEntities: repo,
Executors: repo, Environments: repo, TaskEnvironments: repo,
Reviews: repo,
}, nil, log, service.RepositoryDiscoveryConfig{})
h := &TaskHandlers{service: svc, logger: log}
msg, err := ws.NewRequest("msg-1", ws.ActionTaskCreate, map[string]any{
"workspace_id": "ws-1",
"workflow_id": "wf-1",
"title": "Full review step",
})
require.NoError(t, err)
resp, err := h.wsCreateTask(context.Background(), msg)
require.NoError(t, err)
require.Equal(t, ws.MessageTypeError, resp.Type)
var payload ws.ErrorPayload
require.NoError(t, json.Unmarshal(resp.Payload, &payload))
assert.Equal(t, ws.ErrorCodeConflict, payload.Code)
assert.Contains(t, payload.Message, "WIP limit")
}
func TestWSCreateTaskReturnsValidationErrorForStepOutsideWorkflow(t *testing.T) {
log := newTestLogger(t)
repo := &captureCreateTaskRepo{}
@@ -171,6 +171,9 @@ func (h *TaskHandlers) wsCreateTask(ctx context.Context, msg *ws.Message) (*ws.M
})
if err != nil {
h.logger.Error("failed to create task", zap.Error(err))
if errors.Is(err, service.ErrWIPLimitExceeded) {
return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeConflict, err.Error(), nil)
}
if isTaskCreateValidationError(err) {
return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, err.Error(), nil)
}
@@ -7,6 +7,7 @@ import (
agentdto "github.com/kandev/kandev/internal/agent/dto"
"github.com/kandev/kandev/internal/task/models"
"github.com/kandev/kandev/internal/task/repository/repoerrors"
wfmodels "github.com/kandev/kandev/internal/workflow/models"
v1 "github.com/kandev/kandev/pkg/api/v1"
)
@@ -15,6 +16,7 @@ var ErrWorkspaceNotFound = repoerrors.ErrWorkspaceNotFound
var ErrTaskNotFound = repoerrors.ErrTaskNotFound
var ErrTaskPlanNotFound = repoerrors.ErrTaskPlanNotFound
var ErrRepositoryNotFound = repoerrors.ErrRepositoryNotFound
var ErrWIPLimitExceeded = wfmodels.ErrWIPLimitExceeded
// WorkspaceRepository handles workspace CRUD.
type WorkspaceRepository interface {
@@ -16,6 +16,7 @@ import (
"github.com/kandev/kandev/internal/db/dialect"
"github.com/kandev/kandev/internal/task/models"
usermodels "github.com/kandev/kandev/internal/user/models"
wfmodels "github.com/kandev/kandev/internal/workflow/models"
v1 "github.com/kandev/kandev/pkg/api/v1"
)
@@ -139,6 +140,20 @@ func runnerProjection(alias string) string {
// (ADR 0005 Wave F); when the request carries AssigneeAgentProfileID we
// upsert a 'runner' row in workflow_step_participants instead.
func (r *Repository) CreateTask(ctx context.Context, task *models.Task) error {
return r.createTask(ctx, task, "", 0)
}
// CreateTaskIfWorkflowStepHasCapacity atomically admits a task into a
// WIP-limited workflow step. The occupancy check and insert share one writer
// transaction, so concurrent watcher events cannot overfill the step.
func (r *Repository) CreateTaskIfWorkflowStepHasCapacity(ctx context.Context, task *models.Task, targetStepID string, limit int) error {
if limit <= 0 {
return r.CreateTask(ctx, task)
}
return r.createTask(ctx, task, targetStepID, limit)
}
func (r *Repository) createTask(ctx context.Context, task *models.Task, targetStepID string, limit int) error {
if task.ID == "" {
task.ID = uuid.New().String()
}
@@ -163,6 +178,11 @@ func (r *Repository) CreateTask(ctx context.Context, task *models.Task) error {
if err != nil {
return err
}
defer func() { _ = tx.Rollback() }()
if err := r.ensureWorkflowStepCapacity(ctx, tx, targetStepID, limit); err != nil {
return err
}
_, err = tx.ExecContext(ctx, r.db.Rebind(`
INSERT INTO tasks (id, workspace_id, workflow_id, workflow_step_id, title, description, state, priority, position, metadata, is_ephemeral, parent_id, created_at, updated_at, origin, project_id, labels, identifier)
@@ -187,6 +207,36 @@ func (r *Repository) CreateTask(ctx context.Context, task *models.Task) error {
return tx.Commit()
}
func (r *Repository) ensureWorkflowStepCapacity(ctx context.Context, tx *sql.Tx, targetStepID string, limit int) error {
if limit <= 0 {
return nil
}
if err := lockWorkflowStepForCapacity(ctx, tx, r.db.DriverName(), r.db.Rebind, targetStepID); err != nil {
return err
}
var occupants int
if err := tx.QueryRowContext(ctx, r.db.Rebind(`
SELECT COUNT(*) FROM tasks
WHERE workflow_step_id = ?
AND archived_at IS NULL
AND is_ephemeral = 0
`), targetStepID).Scan(&occupants); err != nil {
return err
}
if occupants >= limit {
return wfmodels.NewWIPLimitError(targetStepID, limit, occupants)
}
return nil
}
func lockWorkflowStepForCapacity(ctx context.Context, tx *sql.Tx, driver string, rebind func(string) string, stepID string) error {
if !dialect.IsPostgres(driver) {
return nil
}
var lockedID string
return tx.QueryRowContext(ctx, rebind(`SELECT id FROM workflow_steps WHERE id = ? FOR UPDATE`), stepID).Scan(&lockedID)
}
// upsertRunnerInTx writes (or replaces) a 'runner' participant row for
// (stepID, taskID) inside the provided transaction. Mirrors
// workflow.Repository.SetTaskRunner but reuses the caller's tx.
@@ -367,6 +417,10 @@ func (r *Repository) UpdateTaskIfWorkflowStepHasCapacity(ctx context.Context, ta
}
defer func() { _ = tx.Rollback() }()
if err := lockWorkflowStepForCapacity(ctx, tx, r.db.DriverName(), r.db.Rebind, targetStepID); err != nil {
return err
}
var occupants int
if err := tx.QueryRowContext(ctx, r.db.Rebind(`
SELECT COUNT(*) FROM tasks
@@ -378,7 +432,7 @@ func (r *Repository) UpdateTaskIfWorkflowStepHasCapacity(ctx context.Context, ta
return err
}
if occupants >= limit {
return fmt.Errorf("WIP limit exceeded for workflow step %s: limit %d already occupied", targetStepID, limit)
return wfmodels.NewWIPLimitError(targetStepID, limit, occupants)
}
result, err := tx.ExecContext(ctx, r.db.Rebind(`
@@ -0,0 +1,99 @@
package repository
import (
"context"
"errors"
"fmt"
"strings"
"sync"
"testing"
"github.com/kandev/kandev/internal/task/models"
wfmodels "github.com/kandev/kandev/internal/workflow/models"
v1 "github.com/kandev/kandev/pkg/api/v1"
)
type workflowStepCapacityCreator interface {
CreateTaskIfWorkflowStepHasCapacity(context.Context, *models.Task, string, int) error
}
func TestUpdateTaskIfWorkflowStepHasCapacity_ReturnsTypedWIPError(t *testing.T) {
repo, cleanup := createTestSQLiteRepo(t)
defer cleanup()
ctx := context.Background()
if err := repo.CreateTask(ctx, &models.Task{
ID: "wip-existing", WorkspaceID: "wip-workspace", WorkflowID: "wip-workflow",
WorkflowStepID: "wip-step", Title: "Existing", State: v1.TaskStateCreated,
}); err != nil {
t.Fatalf("seed existing task: %v", err)
}
candidate := &models.Task{
ID: "wip-candidate", WorkspaceID: "wip-workspace", WorkflowID: "wip-workflow",
WorkflowStepID: "other-step", Title: "Candidate", State: v1.TaskStateCreated,
}
err := repo.UpdateTaskIfWorkflowStepHasCapacity(ctx, candidate, "wip-step", "wip-candidate", 1)
if err == nil || !errors.Is(err, wfmodels.ErrWIPLimitExceeded) {
t.Fatalf("error=%v, want typed WIP limit error", err)
}
}
func TestCreateTaskIfWorkflowStepHasCapacity_Concurrent(t *testing.T) {
repo, cleanup := createTestSQLiteRepo(t)
defer cleanup()
creator, ok := any(repo).(workflowStepCapacityCreator)
if !ok {
t.Fatal("task repository does not implement atomic workflow-step capacity creation")
}
const (
workerCount = 8
stepID = "wip-step"
)
ctx := context.Background()
start := make(chan struct{})
results := make(chan error, workerCount)
var wg sync.WaitGroup
for i := 0; i < workerCount; i++ {
wg.Add(1)
go func(index int) {
defer wg.Done()
<-start
results <- creator.CreateTaskIfWorkflowStepHasCapacity(ctx, &models.Task{
ID: fmt.Sprintf("wip-task-%d", index),
WorkspaceID: "wip-workspace",
WorkflowID: "wip-workflow",
WorkflowStepID: stepID,
Title: fmt.Sprintf("Task %d", index),
State: v1.TaskStateCreated,
}, stepID, 2)
}(i)
}
close(start)
wg.Wait()
close(results)
created, rejected := 0, 0
for err := range results {
if err == nil {
created++
continue
}
if !strings.Contains(strings.ToLower(err.Error()), "wip limit exceeded") {
t.Fatalf("unexpected create error: %v", err)
}
rejected++
}
if created != 2 || rejected != workerCount-2 {
t.Fatalf("created=%d rejected=%d, want created=2 rejected=%d", created, rejected, workerCount-2)
}
occupants, err := repo.CountTasksByWorkflowStep(ctx, stepID)
if err != nil {
t.Fatalf("count occupants: %v", err)
}
if occupants != 2 {
t.Fatalf("occupants=%d, want 2", occupants)
}
}
@@ -145,6 +145,7 @@ type StartStepResolver interface {
var (
ErrActiveTaskSessions = errors.New("active agent sessions exist")
ErrWIPLimitExceeded = wfmodels.ErrWIPLimitExceeded
ErrInvalidRepositorySettings = errors.New("invalid repository settings")
ErrInvalidExecutorConfig = errors.New("invalid executor config")
// Workspace-source sentinels are the service boundary consumed by the HTTP
@@ -79,6 +79,10 @@ type taskEnvironmentOwnerTransferer interface {
TransferTaskEnvironmentToTask(ctx context.Context, envID, taskID string) error
}
type workflowStepCapacityTaskCreator interface {
CreateTaskIfWorkflowStepHasCapacity(ctx context.Context, task *models.Task, targetStepID string, limit int) error
}
// Task operations
// isOfficeRequest returns true if the request should create an office task.
@@ -133,7 +137,7 @@ func (s *Service) CreateTask(ctx context.Context, req *CreateTaskRequest) (*mode
}
}
if err := s.tasks.CreateTask(ctx, task); err != nil {
if err := s.createTaskWithCapacity(ctx, task); err != nil {
s.logger.Error("failed to create task", zap.Error(err))
return nil, err
}
@@ -163,6 +167,30 @@ func (s *Service) CreateTask(ctx context.Context, req *CreateTaskRequest) (*mode
return task, nil
}
func (s *Service) createTaskWithCapacity(ctx context.Context, task *models.Task) error {
if task.IsEphemeral || task.WorkflowStepID == "" {
return s.tasks.CreateTask(ctx, task)
}
if s.workflowStepGetter == nil {
return s.tasks.CreateTask(ctx, task)
}
step, err := s.workflowStepGetter.GetStep(ctx, task.WorkflowStepID)
if err != nil {
return fmt.Errorf("load workflow step %s for task creation: %w", task.WorkflowStepID, err)
}
if step == nil {
return fmt.Errorf("%w: workflow step not found: %s", ErrInvalidTaskWorkflow, task.WorkflowStepID)
}
if step.WIPLimit <= 0 {
return s.tasks.CreateTask(ctx, task)
}
creator, ok := s.tasks.(workflowStepCapacityTaskCreator)
if !ok {
return fmt.Errorf("workflow step %s has WIP limit %d but capacity-aware task persistence is unavailable", step.ID, step.WIPLimit)
}
return creator.CreateTaskIfWorkflowStepHasCapacity(ctx, task, step.ID, step.WIPLimit)
}
// inheritParentRepositories fills req.Repositories from the parent task when a
// subtask is created without explicit repositories. This applies to any
// repo-less subtask (not only inherit_parent ones), matching the MCP
@@ -0,0 +1,121 @@
package service
import (
"context"
"errors"
"testing"
"github.com/kandev/kandev/internal/task/models"
wfmodels "github.com/kandev/kandev/internal/workflow/models"
v1 "github.com/kandev/kandev/pkg/api/v1"
)
type fixedStartStepResolver struct {
stepID string
}
func (r fixedStartStepResolver) ResolveStartStep(context.Context, string) (string, error) {
return r.stepID, nil
}
func (r fixedStartStepResolver) ResolveFirstStep(context.Context, string) (string, error) {
return r.stepID, nil
}
func TestCreateTask_RejectsFullWIPStepBeforePersistence(t *testing.T) {
svc, events, repo := createTestService(t)
ctx := context.Background()
seedWIPWorkflow(t, ctx, repo)
svc.SetWorkflowStepGetter(&fakeWorkflowStepGetter{steps: map[string]*wfmodels.WorkflowStep{
"review-step": {ID: "review-step", WorkflowID: "wip-workflow", Name: "Review", WIPLimit: 1},
}})
if err := repo.CreateTask(ctx, &models.Task{
ID: "wip-occupant", WorkspaceID: "wip-workspace", WorkflowID: "wip-workflow",
WorkflowStepID: "review-step", Title: "Occupant", State: v1.TaskStateCreated,
}); err != nil {
t.Fatalf("seed occupant: %v", err)
}
_, err := svc.CreateTask(ctx, &CreateTaskRequest{
WorkspaceID: "wip-workspace", WorkflowID: "wip-workflow", WorkflowStepID: "review-step",
Title: "Rejected", Description: "must not persist",
})
if err == nil || !errors.Is(err, wfmodels.ErrWIPLimitExceeded) {
t.Fatalf("error=%v, want typed WIP limit rejection", err)
}
if _, err := repo.GetTask(ctx, "wip-occupant"); err != nil {
t.Fatalf("occupant disappeared: %v", err)
}
tasks, err := repo.ListTasksByWorkflowStep(ctx, "review-step")
if err != nil {
t.Fatalf("list step tasks: %v", err)
}
if len(tasks) != 1 {
t.Fatalf("step task count=%d, want 1", len(tasks))
}
if len(events.GetPublishedEvents()) != 0 {
t.Fatalf("published events=%d, want none", len(events.GetPublishedEvents()))
}
}
func TestCreateTask_ResolvedStartStepUsesWIPAdmission(t *testing.T) {
svc, _, repo := createTestService(t)
ctx := context.Background()
seedWIPWorkflow(t, ctx, repo)
svc.SetWorkflowStepGetter(&fakeWorkflowStepGetter{steps: map[string]*wfmodels.WorkflowStep{
"review-step": {ID: "review-step", WorkflowID: "wip-workflow", Name: "Review", WIPLimit: 1},
}})
svc.SetStartStepResolver(fixedStartStepResolver{stepID: "review-step"})
if err := repo.CreateTask(ctx, &models.Task{
ID: "wip-occupant-resolved", WorkspaceID: "wip-workspace", WorkflowID: "wip-workflow",
WorkflowStepID: "review-step", Title: "Occupant", State: v1.TaskStateCreated,
}); err != nil {
t.Fatalf("seed occupant: %v", err)
}
_, err := svc.CreateTask(ctx, &CreateTaskRequest{
WorkspaceID: "wip-workspace", WorkflowID: "wip-workflow",
Title: "Rejected resolved start", Description: "must not persist",
})
if err == nil || !errors.Is(err, wfmodels.ErrWIPLimitExceeded) {
t.Fatalf("error=%v, want typed WIP limit rejection", err)
}
}
func TestCreateTask_UnlimitedWIPStepPreservesCreation(t *testing.T) {
svc, _, repo := createTestService(t)
ctx := context.Background()
seedWIPWorkflow(t, ctx, repo)
svc.SetWorkflowStepGetter(&fakeWorkflowStepGetter{steps: map[string]*wfmodels.WorkflowStep{
"unlimited-step": {ID: "unlimited-step", WorkflowID: "wip-workflow", Name: "Unlimited"},
}})
for i := 0; i < 3; i++ {
if _, err := svc.CreateTask(ctx, &CreateTaskRequest{
WorkspaceID: "wip-workspace", WorkflowID: "wip-workflow", WorkflowStepID: "unlimited-step",
Title: "Unlimited task", Description: "allowed",
}); err != nil {
t.Fatalf("create task %d: %v", i, err)
}
}
occupants, err := repo.CountTasksByWorkflowStep(ctx, "unlimited-step")
if err != nil {
t.Fatalf("count occupants: %v", err)
}
if occupants != 3 {
t.Fatalf("occupants=%d, want 3", occupants)
}
}
func seedWIPWorkflow(t *testing.T, ctx context.Context, repo interface {
CreateWorkspace(context.Context, *models.Workspace) error
CreateWorkflow(context.Context, *models.Workflow) error
}) {
t.Helper()
if err := repo.CreateWorkspace(ctx, &models.Workspace{ID: "wip-workspace", Name: "WIP Workspace"}); err != nil {
t.Fatalf("seed workspace: %v", err)
}
if err := repo.CreateWorkflow(ctx, &models.Workflow{ID: "wip-workflow", WorkspaceID: "wip-workspace", Name: "WIP Workflow"}); err != nil {
t.Fatalf("seed workflow: %v", err)
}
}
@@ -677,7 +677,7 @@ func (s *Service) validateMoveWIPLimit(ctx context.Context, task *models.Task, t
return fmt.Errorf("failed to count target workflow step tasks: %w", err)
}
if occupants >= targetStep.WIPLimit {
return fmt.Errorf("WIP limit exceeded for workflow step %s: limit %d already occupied", targetStep.ID, targetStep.WIPLimit)
return wfmodels.NewWIPLimitError(targetStep.ID, targetStep.WIPLimit, occupants)
}
return nil
}
@@ -839,8 +839,7 @@ func (s *Service) validateBulkMoveWIPCapacity(ctx context.Context, tasks []*mode
return fmt.Errorf("failed to count target workflow step tasks: %w", err)
}
if occupants+incoming > targetStep.WIPLimit {
return fmt.Errorf("WIP limit exceeded for workflow step %s: limit %d, occupied %d, moving %d",
targetStep.ID, targetStep.WIPLimit, occupants, incoming)
return fmt.Errorf("%w: moving %d task(s)", wfmodels.NewWIPLimitError(targetStep.ID, targetStep.WIPLimit, occupants), incoming)
}
return nil
}
@@ -0,0 +1,30 @@
package models
import (
"errors"
"fmt"
)
// ErrWIPLimitExceeded identifies a workflow-step capacity rejection.
var ErrWIPLimitExceeded = errors.New("workflow step WIP limit exceeded")
// WIPLimitError carries workflow-step capacity details while preserving
// errors.Is(err, ErrWIPLimitExceeded) classification at service boundaries.
type WIPLimitError struct {
StepID string
Limit int
Occupied int
}
func (e *WIPLimitError) Error() string {
return fmt.Sprintf("WIP limit exceeded for workflow step %s: limit %d already occupied (%d)", e.StepID, e.Limit, e.Occupied)
}
func (e *WIPLimitError) Unwrap() error {
return ErrWIPLimitExceeded
}
// NewWIPLimitError constructs a typed capacity conflict for a workflow step.
func NewWIPLimitError(stepID string, limit, occupied int) error {
return &WIPLimitError{StepID: stepID, Limit: limit, Occupied: occupied}
}
@@ -574,7 +574,6 @@ test.describe("Mobile kanban view", () => {
is_start_step: true,
});
await apiClient.createWorkflowStep(workflow.id, "Done", 1);
await apiClient.updateWorkflowStep(limitedStep.id, { wip_limit: 1 });
await apiClient.createTask(seedData.workspaceId, "Mobile WIP One", {
workflow_id: workflow.id,
workflow_step_id: limitedStep.id,
@@ -583,6 +582,9 @@ test.describe("Mobile kanban view", () => {
workflow_id: workflow.id,
workflow_step_id: limitedStep.id,
});
// Seed a legacy over-limit state by applying the limit after task creation;
// creation itself must now reject capacity overflow.
await apiClient.updateWorkflowStep(limitedStep.id, { wip_limit: 1 });
await apiClient.saveUserSettings({
workspace_id: seedData.workspaceId,
workflow_filter_id: workflow.id,
@@ -0,0 +1,199 @@
---
spec: docs/specs/tasks/wip-limit-pull-system.md
created: 2026-07-27
status: complete
---
# Implementation Plan: WIP-Limited Task Creation
## Overview
Task moves already enforce workflow-step WIP limits, but `CreateTask` inserts a
task directly into its resolved step without capacity admission. GitHub review
watches fan out one goroutine per newly observed PR and then auto-start every
successfully created task, so a WIP-2 start step can receive and start all eight
tasks from one poll.
The fix adds a typed, transactionally enforced WIP admission primitive, uses it
from task creation for both explicit and resolved start steps, maps capacity
rejections consistently across HTTP, WebSocket, MCP, and watcher dispatch, and
documents the corrected behavior. Watcher integration coverage also locks in
the existing lifecycle contract: an admitted task remains in its configured
step while the agent is starting and working, and advances only on a genuine
turn-complete event.
## Confirmed Root Cause
- `github.Service.TriggerReviewWatch` publishes an event for every new PR.
- `orchestrator.Service.handleNewReviewPR` starts one background
`createReviewTask` goroutine per event.
- `task/service.Service.CreateTask` calls `TaskRepository.CreateTask` without
reading the resolved step's `WIPLimit`.
- WIP validation exists only on move/transition paths.
- `createReviewTask` calls `StartTask` for every task created in an
`auto_start_agent` step.
The smallest reproduction is an empty auto-start workflow start step with
`wip_limit: 2` receiving eight concurrent non-ephemeral task creations. Current
behavior persists eight tasks; the required behavior admits exactly two.
---
## Backend
### Typed WIP capacity contract
- Add `ErrWIPLimitExceeded` to
`apps/backend/internal/workflow/models/errors.go` so task repository, task
service, HTTP/WS/MCP adapters, and orchestrator code classify capacity
without parsing error strings.
- Preserve user-facing context by wrapping the sentinel with the step ID,
configured limit, and current occupancy.
### Atomic repository admission
- Extend the narrow task-placement repository contract with
`CreateTaskIfWorkflowStepHasCapacity(ctx, task, targetStepID, limit)`.
- Implement it in
`apps/backend/internal/task/repository/sqlite/task.go` using the same task
insert and runner-participant transaction as `CreateTask`.
- Serialize admission on the target `workflow_steps` row. PostgreSQL uses
`SELECT ... FOR UPDATE`; SQLite already uses a single writer connection, so
the count and insert remain in one serialized write transaction.
- Count only non-archived, non-ephemeral occupants, matching existing WIP move
semantics.
- Reuse the step-capacity lock and typed sentinel from
`UpdateTaskIfWorkflowStepHasCapacity` so creation, moves, and pulls share one
concurrency rule.
### Task service enforcement
- In `task/service.Service.CreateTask`, resolve and validate the final workflow
step before persistence.
- For a positive `WIPLimit`, call the capacity-aware repository method instead
of the unconditional insert. This includes requests that omit
`workflow_step_id` and resolve to the workflow start step.
- Fail closed when a positive WIP limit is configured but the repository does
not implement capacity-aware creation.
- Leave `wip_limit: 0`, workflow-less ephemeral tasks, and existing unlimited
task creation unchanged.
- Ensure a capacity rejection occurs before blockers, repositories,
`task.created`, task sessions, or auto-start side effects.
### Caller error behavior
- HTTP task creation returns `409 Conflict` with the WIP error.
- WebSocket and MCP task creation return `ErrorCodeConflict` with the same
message.
- GitHub review dispatch recognizes `ErrWIPLimitExceeded` as a deferral:
release the review-PR reservation, do not attach a task ID or auto-start, and
log at a non-error level so later polls can retry without false failure
noise.
- Other task-creation errors retain their current handling.
---
## Tests
- **What:** concurrent repository admission never exceeds WIP capacity.
**File:** `apps/backend/internal/task/repository/task_wip_test.go`.
**How:** seed a WIP-2 step, synchronize eight goroutines, assert exactly two
inserts and six `errors.Is(err, ErrWIPLimitExceeded)` results; run with
`-race`.
- **What:** explicit and resolved start-step creation obey WIP, while
`wip_limit: 0` remains unlimited and rejected creation emits no
`task.created`.
**File:** new
`apps/backend/internal/task/service/service_tasks_wip_test.go`.
**How:** SQLite-backed service integration tests using real workflow/task
repositories.
- **What:** HTTP, WebSocket, and MCP task creation classify WIP rejection as a
conflict.
**Files:** task handler and MCP handler tests.
**How:** focused adapter tests asserting status/error code and message.
- **What:** review-watch capacity rejection releases the PR reservation and
never assigns a task ID or auto-starts.
**File:**
`apps/backend/internal/orchestrator/event_handlers_github_review_test.go`.
**How:** deterministic fake creator returning the typed WIP error, followed
by a retry that succeeds.
- **What:** an admitted watcher task does not advance its workflow during
startup or active work.
**File:**
`apps/backend/internal/orchestrator/event_handlers_github_review_test.go`.
**How:** create directly in an auto-start `Review` step, assert the task
remains there after creation and boot-ready, then emit a real turn-complete
event and assert one transition to `Done`.
## Frontend
No frontend code changes are required. Existing task-creation surfaces already
display backend errors; the backend adapters must return a conflict instead of
an internal error.
## E2E Tests
No browser E2E is planned because there is no visual or interaction change.
The concurrency invariant and every external error boundary are covered by
targeted Go tests.
## Public Documentation
- Update `docs/public/tasks-and-workflows.md` and
`docs/public/workflow-tips.md` to state that WIP applies to initial task
creation, including integration-created tasks, and that full-step creation
is rejected for later retry.
- Validate the public docs after the wording changes.
## Implementation Waves And Parallel Candidates
All tasks are sequential because each consumes the typed contract or behavior
introduced by the preceding task.
- [x] [Task 01: Atomic repository admission](task-01-atomic-repository-admission.md)
- [x] [Task 02: Task-service WIP enforcement](task-02-task-service-enforcement.md)
- [x] [Task 03: Conflict adapters and review-watch deferral](task-03-conflict-adapters-and-watcher-deferral.md)
- [x] [Task 04: Public documentation](task-04-public-documentation.md)
## Completion Evidence
- Repository admission is transactionally serialized, with PostgreSQL step-row
locking and SQLite writer serialization; focused repository tests pass.
- Task creation enforces explicit and resolved workflow-step WIP limits, while
unlimited and ephemeral creation remain compatible; focused service tests
pass.
- HTTP returns `409 Conflict`; WebSocket and MCP return conflict error codes;
watcher capacity failures release reservations and skip assignment/start.
- Review watcher lifecycle coverage confirms boot-ready does not advance the
task and the first genuine turn completion moves Review to Done once.
- Public documentation validation: `node scripts/validate-public-docs.test.mjs`
(58/58 passed).
- Backend verification: `go test -tags fts5 ./internal/task/repository
./internal/task/service ./internal/task/handlers ./internal/mcp/handlers
./internal/orchestrator -count=1` (2,562 tests passed).
No task is marked `parallel-safe`; waves do not authorize subagent execution.
## Risks
- A pre-insert count outside the write transaction would preserve the original
race. Admission must be serialized per target step in both SQLite and
PostgreSQL.
- Rejected review PRs must release their dedup reservations or they will never
retry.
- Watcher auto-start coverage must distinguish `agent.boot_ready` from a real
turn-complete event; treating startup readiness as completion would move a
newly admitted task out of the limited step before its review runs.
- The resolved start-step path must be covered explicitly; checking only
request-supplied `workflow_step_id` would leave the reported configuration
broken.
- Existing tests intentionally seed over-limit UI states through task creation.
Those fixtures must use repository-level setup or another explicit legacy
state mechanism rather than weakening the production invariant.
## Out of Scope
- A GitHub review-watch `max_inflight_tasks` setting.
- Profile-wide session concurrency limits.
- Automatic feeder-step creation or watcher configuration changes.
- UI redesign or new WIP controls.
@@ -0,0 +1,65 @@
---
id: "01-atomic-repository-admission"
title: "Atomic repository admission"
status: done
wave: 1
depends_on: []
plan: "plan.md"
spec: "../../specs/tasks/wip-limit-pull-system.md"
---
# Task 01: Atomic Repository Admission
## Acceptance
- The task repository can create a task only when the target workflow step has
capacity, counting active non-ephemeral occupants with the same semantics as
WIP-limited moves.
- Eight synchronized creates against an empty WIP-2 step persist exactly two
tasks; every rejected call matches the typed WIP sentinel.
- PostgreSQL admission locks the target workflow-step row for the transaction;
SQLite admission remains serialized through its single writer.
- Existing capacity-aware updates return the same typed WIP error.
## Verification
```bash
cd apps/backend
go test -tags fts5 -race ./internal/task/repository/... -run 'Test.*WorkflowStep.*Capacity|Test.*WIP.*Concurrent' -count=1
```
## Files likely touched
- `apps/backend/internal/workflow/models/errors.go`
- `apps/backend/internal/task/repository/interface.go`
- `apps/backend/internal/task/repository/sqlite/task.go`
- `apps/backend/internal/task/repository/task_repository_test.go`
- `apps/backend/internal/task/repository/sqlite/task_order_test.go` or a focused
dialect test if the PostgreSQL lock query is extracted
## Dependencies
None.
## Parallelism
`sequential`
## Inputs
- Spec sections `What`, `Failure Modes`, and `Scenarios`.
- Existing `UpdateTaskIfWorkflowStepHasCapacity` transaction.
- SQLite writer configuration in `apps/backend/internal/db/sqlite.go`.
- PostgreSQL dialect detection in `apps/backend/internal/db/dialect`.
## Output contract
Mark this task `in_progress` before the RED test and `done` after GREEN and
refactor. Update `plan.md` in the same conversation and report the failing-test
evidence, repository API/transaction changes, files changed, exact test result,
and remaining database portability risks.
## Evidence
`go test ./internal/task/repository -run 'Test(CreateTaskIfWorkflowStepHasCapacity|UpdateTaskIfWorkflowStepHasCapacity)' -count=1`
passed (2 tests).
@@ -0,0 +1,63 @@
---
id: "02-task-service-enforcement"
title: "Task-service WIP enforcement"
status: done
wave: 2
depends_on: ["01-atomic-repository-admission"]
plan: "plan.md"
spec: "../../specs/tasks/wip-limit-pull-system.md"
---
# Task 02: Task-Service WIP Enforcement
## Acceptance
- `Service.CreateTask` uses atomic capacity admission for a positive WIP limit
after resolving the final workflow step.
- Explicit and automatically resolved start-step creation reject a full step
with the typed WIP error and persist no task, repository, blocker, event, or
session side effect.
- Unlimited workflow steps and workflow-less ephemeral tasks preserve current
behavior.
- A positive WIP limit fails closed if capacity-aware persistence is
unavailable.
## Verification
```bash
cd apps/backend
go test -tags fts5 ./internal/task/service -run 'TestCreateTask.*WIP|TestCreateTask.*StartStep|TestCreateTask.*Unlimited' -count=1
```
## Files likely touched
- `apps/backend/internal/task/service/service_tasks.go`
- `apps/backend/internal/task/service/service_tasks_wip_test.go`
- `apps/backend/internal/task/service/service.go` only if a narrow shared
interface or helper belongs there
## Dependencies
Task 01.
## Parallelism
`sequential`
## Inputs
- Task 01's typed error and capacity-aware repository method.
- `resolveWorkflowStep`, `validateTaskWorkflow`, and `buildTask` in
`service_tasks.go`.
- Existing task-event test helpers in `service_events_test.go`.
## Output contract
Mark this task `in_progress` before the RED test and `done` after GREEN and
refactor. Update `plan.md` and report explicit/resolved-step coverage,
side-effect assertions, files changed, exact test result, blockers, and risks.
## Evidence
Explicit-step, resolved-start-step, unlimited, and no-event rejection tests
pass in `internal/task/service`.
@@ -0,0 +1,78 @@
---
id: "03-conflict-adapters-and-watcher-deferral"
title: "Conflict adapters and review-watch deferral"
status: done
wave: 3
depends_on: ["02-task-service-enforcement"]
plan: "plan.md"
spec: "../../specs/tasks/wip-limit-pull-system.md"
---
# Task 03: Conflict Adapters and Review-Watch Deferral
## Acceptance
- HTTP task creation returns `409 Conflict` for the typed WIP error.
- WebSocket and MCP task creation return `ErrorCodeConflict` and preserve the
actionable step/limit message.
- GitHub review task creation treats WIP rejection as deferred capacity:
releases the PR reservation, assigns no task ID, starts no task, and permits a
later retry to succeed.
- An admitted GitHub review task created directly in an auto-start `Review`
step remains in that step during startup and active work; `agent.boot_ready`
does not advance it, and the first genuine turn completion moves it to the
configured next step exactly once.
- Non-capacity task-creation failures retain their current error behavior.
## Verification
```bash
cd apps/backend
go test -tags fts5 ./internal/task/handlers -run 'Test.*CreateTask.*WIP' -count=1
go test -tags fts5 ./internal/mcp/handlers -run 'TestHandleCreateTask.*WIP' -count=1
go test -tags fts5 ./internal/orchestrator -run 'Test(CreateReviewTask_.*|GitHubWatcherTaskCreateError|ReviewWatchLifecycle_.*)' -count=1
```
## Files likely touched
- `apps/backend/internal/task/handlers/errors.go`
- `apps/backend/internal/task/handlers/task_http_handlers.go`
- `apps/backend/internal/task/handlers/task_http_handlers_test.go`
- `apps/backend/internal/task/handlers/task_ws_handlers.go`
- `apps/backend/internal/task/handlers/task_ws_handlers_test.go`
- `apps/backend/internal/mcp/handlers/handlers.go`
- `apps/backend/internal/mcp/handlers/handlers_test.go`
- `apps/backend/internal/orchestrator/event_handlers_github.go`
- `apps/backend/internal/orchestrator/event_handlers_github_review_test.go`
## Dependencies
Task 02.
## Evidence
HTTP, WebSocket, and MCP focused tests pass. GitHub review capacity rejection
coverage confirms reservation release without task assignment or auto-start;
the lifecycle regression confirms boot-ready stays in Review and the first
genuine turn completion advances once.
## Parallelism
`sequential`
## Inputs
- Task 02's service error behavior.
- Existing `isMoveConflict`, `isTaskCreateValidationError`, and WebSocket error
mappings.
- Existing review reservation release test in
`event_handlers_github_review_test.go`.
- Existing boot-ready/turn-complete separation in
`event_handlers_agent.go` and its regression tests.
## Output contract
Mark this task `in_progress` before the RED tests and `done` after GREEN and
refactor. Update `plan.md` and report each adapter's observed error contract,
review reservation/retry evidence, watcher lifecycle-ordering evidence, files
changed, exact test results, blockers, and risks.
@@ -0,0 +1,56 @@
---
id: "04-public-documentation"
title: "Public WIP documentation"
status: done
wave: 4
depends_on: ["03-conflict-adapters-and-watcher-deferral"]
plan: "plan.md"
spec: "../../specs/tasks/wip-limit-pull-system.md"
---
# Task 04: Public WIP Documentation
## Acceptance
- Public workflow documentation states that WIP capacity applies to initial
task creation as well as moves and transitions.
- Documentation explains that integration-created tasks rejected for capacity
remain deferred for later retry and do not auto-start.
- Public-doc validation succeeds.
## Evidence
Updated `docs/public/tasks-and-workflows.md` and
`docs/public/workflow-tips.md`. `node scripts/validate-public-docs.test.mjs`
passed all 58 tests.
## Verification
```bash
node --test scripts/validate-public-docs.test.mjs
node scripts/validate-public-docs.mjs
```
## Files likely touched
- `docs/public/tasks-and-workflows.md`
- `docs/public/workflow-tips.md`
## Dependencies
Task 03.
## Parallelism
`sequential`
## Inputs
- Amended WIP spec.
- Final typed conflict and review-watch deferral behavior from Tasks 01-03.
## Output contract
Mark this task `in_progress` before editing and `done` after validation. Update
`plan.md` and report the public wording changed, files changed, exact validation
results, blockers, and residual documentation risks.
+19
View File
@@ -182,6 +182,19 @@ New steps allow manual moves by default. **Show in command panel** also defaults
| **WIP limit** | Maximum active, non-archived, non-ephemeral tasks accepted by the step. `0` means unlimited. A move into a full step is rejected; reordering within the same step does not consume another slot. |
| **Pull from** | Optional feeder step. When capacity opens, Kandev pulls eligible work from that step. It requires a nonzero WIP limit. |
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
admission check is atomic so concurrent creates cannot collectively exceed the
limit. A full step returns a conflict that includes the step and limit; the
rejected task, session, and task-created event are not persisted. Ephemeral
tasks are not counted.
Integration watchers use the same admission rule. For example, a GitHub review
watch targeting a `Review` step with a limit of two admits at most two newly
observed pull requests at a time. Pull requests that lose the capacity race
remain eligible for a later poll; Kandev releases their temporary watch
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. 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`), creation time, and ID. A candidate whose move fails—for example because its session is running or starting—is skipped for that pull pass.
@@ -200,6 +213,12 @@ Generic comment, blocker-resolution, approval, heartbeat, budget, and error trig
When **On Turn Complete** moves a task, **Wait for agent completion signal** is available. With it enabled, a bare turn end leaves the task waiting; the agent must call `step_complete_kandev`. The call requires a summary and can include a handoff or blockers. It is idempotent within the step, runs asynchronously, and a user message sent before the transition is applied cancels that pending signal. Without the option, turn end counts as completion.
An auto-started task stays in its current step while the agent session boots and
while its first turn is running. A boot-ready event is not a turn completion.
For example, a review step with `on_enter: auto_start_agent` and
`on_turn_complete: move_to_next` moves to the next step only after the genuine
review turn completes, not during startup.
Plan mode can be disabled when the turn completes and/or when the task exits the step. A step prompt is Markdown and can include `{{task_prompt}}` to insert the original task description.
### Build a human gate
+15
View File
@@ -70,6 +70,21 @@ Choose this for a deliberate multi-pass delivery process. It is excessive for sm
Choose this for a local first-pass review. For repository-provider watch automation, use the relevant integration instead.
### GitHub review watches and WIP limits
A GitHub Review Watch creates each matching pull request in its configured
workflow step. If that step has a WIP limit, admission is atomic: only tasks
that obtain a slot are created. An admitted task auto-starts only when its
configured workflow step enables `auto_start_agent`. A pull request rejected
because the step is full is deferred for a later poll; its temporary watch
reservation is released, so it is not lost or permanently marked as handled.
If the target step has `auto_start_agent` on entry and `move_to_next` on turn
completion, the task remains in the target step during agent startup and active
review. `agent.boot_ready` only makes the session ready; the first real turn
completion performs the configured move once. Use a human-gate transition when
reviews must wait for approval instead of advancing automatically.
## Build a custom workflow
Choose **Add Workflow**, give it a name, select **Custom**, and save it. Expand each step to edit its behavior. Reorder steps by dragging them; transition actions that say “next” or “previous” follow the saved position order.
+53 -8
View File
@@ -1,5 +1,5 @@
---
status: building
status: shipped
---
# WIP Limit Pull System
@@ -11,9 +11,23 @@ Workflow steps can define a work-in-progress limit and an optional feeder step.
and preserves existing behavior. `pull_from_step_id` is empty by default; when it
is set, it must reference another step in the same workflow.
When a limited step is full, manual moves, drag/drop moves, API moves, MCP moves,
bulk moves, and workflow-engine transitions into that step are rejected instead
of overfilling the step. Same-step reordering is allowed.
When a limited step is full, new task creation, manual moves, drag/drop moves,
API moves, MCP moves, bulk moves, and workflow-engine transitions into that
step are rejected instead of overfilling the step. The creation rule applies
whether the caller names the step explicitly or Kandev resolves it as the
workflow's start step. Same-step reordering is allowed.
Capacity admission is atomic per workflow step. Concurrent task creation and
move attempts cannot collectively admit more active, non-archived,
non-ephemeral tasks than the configured limit.
Integration watchers use the same admission rule as interactive and API task
creation. If a GitHub review watch observes more pull requests than its target
step can accept, Kandev creates only the tasks that obtain capacity and
auto-starts those tasks only when the configured step enables
`auto_start_agent`. Pull requests rejected for capacity remain eligible for a
later poll; Kandev does not leave a task, session, repository association, or
permanent review-watch reservation for a rejected attempt.
When a task leaves a limited step that has a feeder configured, Kandev attempts
to pull queued tasks from the feeder into the vacated step until the step reaches
@@ -52,13 +66,44 @@ ID.
Moving into a full limited step returns a conflict with a user-visible message
that includes the target step and limit. Optimistic UI moves must roll back.
Creating a task in a full limited step returns the same conflict classification
and does not persist any part of the rejected task. Internal automation treats
the conflict as deferred work rather than an integration failure.
If a pull attempt races with another actor that fills the slot, the pull attempt
stops without overfilling the target step.
Deleting a step clears any `pull_from_step_id` that points at it.
## Scope Note
## Scenarios
`agent_profiles.max_concurrent_sessions` is related but separate. This feature
adds per-step WIP limits and pull behavior. Kanban enforcement of profile-level
session caps remains tracked separately unless implemented in the same change.
- **GIVEN** a workflow step with `wip_limit: 2` and two active tasks, **WHEN** a
user, API client, MCP client, or integration watcher creates another
non-ephemeral task directly in that step, **THEN** creation is rejected as a
WIP-capacity conflict and no task or session is persisted.
- **GIVEN** a workflow whose limited start step has `wip_limit: 2`, **WHEN** a
caller creates a task without specifying `workflow_step_id` and the start
step is full, **THEN** creation is rejected by the same capacity rule.
- **GIVEN** an empty auto-start step with `wip_limit: 2`, **WHEN** a GitHub
review watch concurrently dispatches eight newly observed pull requests,
**THEN** exactly two review tasks obtain capacity and auto-start, while the
other six pull requests retain no task or reservation and remain eligible for
a later poll.
- **GIVEN** a review watch targets a `Review` step with
`on_enter: auto_start_agent` and `on_turn_complete: move_to_next`, **WHEN** an
admitted review task starts, **THEN** it remains in `Review` throughout agent
startup and the active turn, ignores boot-ready events for workflow
advancement, and moves to the next step exactly once only after the real
agent turn completes.
- **GIVEN** the review-watch scenario above and one accepted review task later
leaves the limited step, **WHEN** the watch polls again, **THEN** one
previously deferred pull request can obtain the newly available capacity.
- **GIVEN** a step with `wip_limit: 0`, **WHEN** callers create tasks
concurrently in that step, **THEN** creation remains unlimited.
## Out of scope
- WIP limits do not replace `agent_profiles.max_concurrent_sessions` or impose
a profile-wide execution budget.
- This change does not add a separate GitHub review-watch
`max_inflight_tasks` setting.