fix(pipeline): parked pipelines stamped completed_at + spawned steps out of order

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()

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 = ?"

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

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)
+	}
+}

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)