From 82a491cb87af2eced504846e9117f0ea436960e7 Mon Sep 17 00:00:00 2001 From: Bruno Bornsztein Date: Fri, 17 Jul 2026 07:07:33 -0500 Subject: [PATCH 1/2] fix(pipeline): don't stamp completed_at on never-started blocked steps MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit UpdateTaskStatus stamped completed_at for every 'blocked' transition, including DAG-waiting pipeline steps that never ran. A freshly-created pipeline flips its non-root steps to 'blocked' (pipeline.go), so all of them were stamped completed_at = created_at — looking finished on the board and feeding false "done" signals to the workflow sweeps that key off completed_at. 'blocked' covers two cases: a step waiting in a DAG (never started) and a task that ran and is now parked awaiting review. Only stamp completed_at for the latter (started_at != nil), preserving the board's order-by-completed_at for genuinely-closed blocked tasks. Co-Authored-By: Claude Opus 4.8 (1M context) --- internal/db/task_events_test.go | 55 +++++++++++++++++++++++++++++++++ internal/db/tasks.go | 12 ++++++- internal/db/tasks_test.go | 8 +++-- 3 files changed, 71 insertions(+), 4 deletions(-) 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 From 9d4d737210ca9bd10bfa9cfc23781987626c66db Mon Sep 17 00:00:00 2001 From: Bruno Bornsztein Date: Fri, 17 Jul 2026 07:18:08 -0500 Subject: [PATCH 2/2] fix(daemon): refuse to spawn a queued task with open blockers MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Defense-in-depth for the workflow DAG invariant: a step must never run before its blockers complete. GetQueuedTasks should only return ready tasks, but a race or stray status flip can leave a step 'queued' while a blocker is still incomplete — we observed a parked (never-executed) pipeline spawn a downstream review/build step whose [Plan] blocker never ran, editing the main checkout out of order. processNextTask now gates every queued task through admitQueuedTask: if the task still has open blockers it is reverted to 'blocked' and logged loudly (rather than run), so the safety-net sweep re-queues it in order once the blocker really finishes — and the anomaly is captured if the underlying race recurs. Co-Authored-By: Claude Opus 4.8 (1M context) --- internal/executor/blocker_guard_test.go | 95 +++++++++++++++++++++++++ internal/executor/executor.go | 38 ++++++++++ 2 files changed, 133 insertions(+) create mode 100644 internal/executor/blocker_guard_test.go 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)