diff --git a/internal/db/task_events_test.go b/internal/db/task_events_test.go index f6499367..e323d341 100644 --- a/internal/db/task_events_test.go +++ b/internal/db/task_events_test.go @@ -274,6 +274,61 @@ func TestUpdateTaskStatusTimestamps(t *testing.T) { } } +func TestUpdateTaskStatusBlockedTimestamps(t *testing.T) { + // A never-started task flipped to 'blocked' is waiting in a DAG (a pipeline + // step staged behind its dependencies). It has NOT completed, so completed_at + // must stay nil — otherwise a freshly-created, never-run step looks finished. + t.Run("DAG-waiting step (never started) leaves completed_at nil", func(t *testing.T) { + database := setupTestDB(t) + defer database.Close() + + task := &Task{Title: "Waiting step", Status: StatusBacklog, Project: "personal"} + if err := database.CreateTask(task); err != nil { + t.Fatalf("create: %v", err) + } + + if err := database.UpdateTaskStatus(task.ID, StatusBlocked); err != nil { + t.Fatalf("block: %v", err) + } + + updated, err := database.GetTask(task.ID) + if err != nil { + t.Fatalf("get: %v", err) + } + if updated.CompletedAt != nil { + t.Errorf("never-started blocked task should have nil completed_at, got %v", updated.CompletedAt.Time) + } + }) + + // A task that actually ran and is then parked in 'blocked' (agent finished its + // turn, awaiting human review) IS settled — the board orders these by + // completed_at, so it must be stamped. + t.Run("review-parked step (started) sets completed_at", func(t *testing.T) { + database := setupTestDB(t) + defer database.Close() + + task := &Task{Title: "Ran then parked", Status: StatusBacklog, Project: "personal"} + if err := database.CreateTask(task); err != nil { + t.Fatalf("create: %v", err) + } + if err := database.UpdateTaskStatus(task.ID, StatusProcessing); err != nil { + t.Fatalf("process: %v", err) + } + + if err := database.UpdateTaskStatus(task.ID, StatusBlocked); err != nil { + t.Fatalf("block: %v", err) + } + + updated, err := database.GetTask(task.ID) + if err != nil { + t.Fatalf("get: %v", err) + } + if updated.CompletedAt == nil { + t.Error("started-then-blocked task should have completed_at set (board sorts by it)") + } + }) +} + func TestCreateTaskEmitsEvent(t *testing.T) { database := setupTestDB(t) defer database.Close() diff --git a/internal/db/tasks.go b/internal/db/tasks.go index 930aaaef..726a70b1 100644 --- a/internal/db/tasks.go +++ b/internal/db/tasks.go @@ -688,8 +688,18 @@ func (db *DB) UpdateTaskStatus(id int64, status string) error { switch status { case StatusProcessing: query += ", started_at = CURRENT_TIMESTAMP" - case StatusDone, StatusBlocked, StatusArchived: + case StatusDone, StatusArchived: query += ", completed_at = CURRENT_TIMESTAMP" + case StatusBlocked: + // 'blocked' covers two very different cases: a step waiting in a DAG (a + // pipeline step staged behind its dependencies — never started) and a task + // that actually ran and is now parked awaiting human review. Only the latter + // has completed a turn. Stamping a never-started step makes it look finished + // on the board and feeds false "done" signals to the workflow sweeps that key + // off completed_at, so only stamp when the task has genuinely started. + if oldTask != nil && oldTask.StartedAt != nil { + query += ", completed_at = CURRENT_TIMESTAMP" + } } query += " WHERE id = ?" diff --git a/internal/db/tasks_test.go b/internal/db/tasks_test.go index ca5b42e9..bf44aafa 100644 --- a/internal/db/tasks_test.go +++ b/internal/db/tasks_test.go @@ -564,9 +564,11 @@ func TestUpdateTaskStatus(t *testing.T) { if retrieved.Status != StatusBlocked { t.Errorf("expected status %q, got %q", StatusBlocked, retrieved.Status) } - // Blocked sets completed_at - if retrieved.CompletedAt == nil { - t.Error("expected completed_at to be set for blocked status") + // Blocking a task that never started (never went through 'processing') must + // NOT stamp completed_at — it's waiting, not finished. Only a task that + // actually ran and is then parked in 'blocked' gets completed_at. + if retrieved.CompletedAt != nil { + t.Error("expected completed_at to stay nil for a never-started blocked task") } // Change back to backlog diff --git a/internal/executor/blocker_guard_test.go b/internal/executor/blocker_guard_test.go new file mode 100644 index 00000000..1a448def --- /dev/null +++ b/internal/executor/blocker_guard_test.go @@ -0,0 +1,95 @@ +package executor + +import ( + "os" + "testing" + + "github.com/bborn/workflow/internal/config" + "github.com/bborn/workflow/internal/db" +) + +// newGuardTestDB opens a throwaway DB with a project for the guard tests. +func newGuardTestDB(t *testing.T) *db.DB { + t.Helper() + f, err := os.CreateTemp("", "guard-*.db") + if err != nil { + t.Fatal(err) + } + f.Close() + t.Cleanup(func() { os.Remove(f.Name()) }) + + database, err := db.Open(f.Name()) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { database.Close() }) + + if err := database.CreateProject(&db.Project{Name: "test", Path: "/tmp/test"}); err != nil { + t.Fatal(err) + } + return database +} + +// The DAG invariant, enforced at the daemon's spawn gate: a queued step that +// still has an incomplete blocker must never start. If it was mis-queued (a +// race, a stray flip), the daemon refuses it and reverts it to 'blocked' rather +// than running work out of order. +func TestAdmitQueuedTaskRefusesOpenBlocker(t *testing.T) { + database := newGuardTestDB(t) + exec := New(database, &config.Config{}) + + blocker := &db.Task{Title: "blocker", Status: db.StatusBlocked, Project: "test"} + if err := database.CreateTask(blocker); err != nil { + t.Fatal(err) + } + step := &db.Task{Title: "mis-queued step", Status: db.StatusQueued, Project: "test"} + if err := database.CreateTask(step); err != nil { + t.Fatal(err) + } + if err := database.AddDependency(blocker.ID, step.ID, true); err != nil { + t.Fatal(err) + } + + if exec.admitQueuedTask(step) { + t.Error("daemon must refuse a queued step whose blocker is not done") + } + + got, err := database.GetTask(step.ID) + if err != nil { + t.Fatal(err) + } + if got.Status != db.StatusBlocked { + t.Errorf("mis-queued step should be reverted to blocked, got %q", got.Status) + } +} + +// A genuinely-ready queued task (no blockers, or all blockers done) is admitted +// and left queued for the executor to run. +func TestAdmitQueuedTaskAllowsReadyTask(t *testing.T) { + database := newGuardTestDB(t) + exec := New(database, &config.Config{}) + + blocker := &db.Task{Title: "done blocker", Status: db.StatusDone, Project: "test"} + if err := database.CreateTask(blocker); err != nil { + t.Fatal(err) + } + step := &db.Task{Title: "ready step", Status: db.StatusQueued, Project: "test"} + if err := database.CreateTask(step); err != nil { + t.Fatal(err) + } + if err := database.AddDependency(blocker.ID, step.ID, true); err != nil { + t.Fatal(err) + } + + if !exec.admitQueuedTask(step) { + t.Error("daemon must admit a queued step whose blockers are all done") + } + + got, err := database.GetTask(step.ID) + if err != nil { + t.Fatal(err) + } + if got.Status != db.StatusQueued { + t.Errorf("ready step should stay queued, got %q", got.Status) + } +} diff --git a/internal/executor/executor.go b/internal/executor/executor.go index cbc89192..a8bfae5c 100644 --- a/internal/executor/executor.go +++ b/internal/executor/executor.go @@ -1660,6 +1660,14 @@ func (e *Executor) processNextTask(ctx context.Context) { } for _, task := range tasks { + // DAG invariant, last line of defense: never start a task that still has + // an incomplete blocker. A queued task should already be ready, but a race + // or a stray flip can mis-queue a blocked step; admitQueuedTask reverts any + // such task to 'blocked' and logs it, instead of running work out of order. + if !e.admitQueuedTask(task) { + continue + } + // Atomically check-and-set to prevent race where two ticks // both see the task as not-running and spawn duplicate goroutines e.mu.Lock() @@ -1674,6 +1682,36 @@ func (e *Executor) processNextTask(ctx context.Context) { } } +// admitQueuedTask enforces the workflow DAG invariant at the point of spawn: a +// task may only run once all its blockers have completed. GetQueuedTasks should +// only ever return genuinely-ready tasks, but this is the last line of defense — +// if a race or a stray status flip left a step 'queued' while a blocker is still +// incomplete, running it would advance the DAG past work that never happened +// (exactly the premature-spawn we saw a parked pipeline hit). Rather than run it, +// revert it to 'blocked' and log loudly, so the safety-net sweep re-queues it in +// order once the blocker really finishes — and so the anomaly is captured if it +// recurs. Returns true only when the task is clear to run. +func (e *Executor) admitQueuedTask(task *db.Task) bool { + open, err := e.db.GetOpenBlockerCount(task.ID) + if err != nil { + // Fail safe: if we can't confirm the task is ready, don't run it. + e.logger.Error("Blocker check failed; refusing to run task", "id", task.ID, "error", err) + return false + } + if open == 0 { + return true + } + + e.logger.Warn("Refusing to run queued task with open blockers — reverting to blocked", + "id", task.ID, "title", task.Title, "open_blockers", open) + e.logLine(task.ID, "system", fmt.Sprintf( + "Refused to start: %d blocker(s) not yet complete. Reverted to blocked; will re-queue when dependencies finish.", open)) + if err := e.updateStatus(task.ID, db.StatusBlocked); err != nil { + e.logger.Error("Failed to revert mis-queued task to blocked", "id", task.ID, "error", err) + } + return false +} + // ExecuteNow runs a task immediately (blocking). func (e *Executor) ExecuteNow(ctx context.Context, taskID int64) error { task, err := e.db.GetTask(taskID)