fix(workflow): queue runs for reviewers added before a task reaches Review (#2969)

Co-authored-by: nova28 <17953305+nova28@users.noreply.github.com>
Co-authored-by: Claude Sonnet 5 <noreply@anthropic.com>
Co-authored-by: Carlos Florencio <carlosmflorencio@gmail.com>

apps/backend/internal/workflow/engine/phase2_callbacks.go

@@ -102,7 +102,12 @@ func (c QueueRunCallback) resolveTarget(
 		if err != nil {
 			return nil, "", err
 		}
-		agentIDs, err := c.resolveParticipantRole(ctx, stepID, taskID, role)
+		var agentIDs []string
+		if taskID == in.State.TaskID {
+			agentIDs, err = c.resolveParticipantRole(ctx, stepID, taskID, in.State.WorkflowID, role)
+		} else {
+			agentIDs, err = c.resolveParticipantRoleStepScoped(ctx, stepID, taskID, role)
+		}
 		return agentIDs, stepID, err
 	case strings.HasPrefix(target, TargetAgentProfile):
 		id := strings.TrimPrefix(target, TargetAgentProfile)
@@ -166,7 +171,37 @@ func (c QueueRunCallback) resolvePrimary(ctx context.Context, taskID, stepID str
 	return []string{id}, nil
 }
 
-func (c QueueRunCallback) resolveParticipantRole(ctx context.Context, stepID, taskID, role string) ([]string, error) {
+func (c QueueRunCallback) resolveParticipantRole(ctx context.Context, stepID, taskID, workflowID, role string) ([]string, error) {
+	if c.Participants == nil {
+		return nil, fmt.Errorf("%w: queue_run target=participant_role requires ParticipantStore", ErrActionNotYetWired)
+	}
+	seats, err := roleSeatsForFanOut(ctx, c.Participants, stepID, taskID, workflowID, role)
+	if err != nil {
+		return nil, fmt.Errorf("queue_run list participants: %w", err)
+	}
+	ids := make([]string, 0, len(seats))
+	for _, p := range seats {
+		ids = append(ids, p.AgentProfileID)
+	}
+	if len(ids) == 0 {
+		return nil, fmt.Errorf("queue_run: no participants with role %q on step %s", role, stepID)
+	}
+	return ids, nil
+}
+
+// resolveParticipantRoleStepScoped resolves a CROSS-task participant_role
+// target — the pre-WO-30 behaviour, unchanged. It must NOT widen to the
+// any-step/any-workflow slate resolveParticipantRole uses: that slate is
+// only safe for the same-task case, where in.State.WorkflowID scopes
+// gatherParticipantSlate's ListTaskParticipantsForWorkflow lookup to the
+// task's actual current workflow. For a cross-task target the engine has no
+// way to resolve the TARGET task's own workflow id here, so an any-step read
+// would return per-task rows stamped on a step from a workflow the target
+// task has since left (see phase2_sqlite.go's ListParticipantsForTaskAnyStep
+// doc comment: overrides must not leak into quorum evaluation for the new
+// workflow). Staying step-scoped to the target task's current step is the
+// only safe read available.
+func (c QueueRunCallback) resolveParticipantRoleStepScoped(ctx context.Context, stepID, taskID, role string) ([]string, error) {
 	if c.Participants == nil {
 		return nil, fmt.Errorf("%w: queue_run target=participant_role requires ParticipantStore", ErrActionNotYetWired)
 	}
@@ -186,6 +221,27 @@ func (c QueueRunCallback) resolveParticipantRole(ctx context.Context, stepID, ta
 	return ids, nil
 }
 
+// roleSeatsForFanOut gathers the same participant population the quorum
+// guard counts (gatherParticipantSlate — per-task rows at any step, unioned
+// with template rows at the evaluating step), filters to role (deliberately
+// NOT DecisionRequired, unlike the guard's own slate — a fan-out wakes every
+// seat in the role, not just decision-required ones), then dedupes via the
+// guard's own canonicalize/collapse so a reviewer present at multiple steps
+// or as both a per-task and template row is woken exactly once.
+func roleSeatsForFanOut(ctx context.Context, store ParticipantStore, stepID, taskID, workflowID, role string) ([]ParticipantInfo, error) {
+	gathered, err := gatherParticipantSlate(ctx, store, stepID, taskID, workflowID)
+	if err != nil {
+		return nil, err
+	}
+	filtered := make([]ParticipantInfo, 0, len(gathered))
+	for _, p := range gathered {
+		if p.Role == role {
+			filtered = append(filtered, p)
+		}
+	}
+	return collapseByRoleAgent(canonicalizeByTaskRoleAgent(filtered, stepID)), nil
+}
+
 func (c QueueRunCallback) resolveCEO(ctx context.Context, taskID string) ([]string, error) {
 	if c.CEOResolver == nil {
 		return nil, fmt.Errorf("%w: queue_run target=workspace.ceo_agent requires CEOAgentResolver", ErrActionNotYetWired)
@@ -332,7 +388,7 @@ func (c ClearDecisionsCallback) Execute(ctx context.Context, in ActionInput) (Ac
 }
 
 // QueueRunForEachParticipantCallback fans out queue_run over every participant
-// on the step matching the configured role.
+// in the task's workflow-scoped participant slate matching the configured role.
 type QueueRunForEachParticipantCallback struct {
 	Adapter      RunQueueAdapter
 	Participants ParticipantStore
@@ -351,15 +407,12 @@ func (c QueueRunForEachParticipantCallback) Execute(ctx context.Context, in Acti
 		return ActionResult{}, fmt.Errorf("queue_run_for_each_participant missing role")
 	}
 	taskID := in.State.TaskID
-	all, err := c.Participants.ListStepParticipants(ctx, in.Step.ID, taskID)
+	seats, err := roleSeatsForFanOut(ctx, c.Participants, in.Step.ID, taskID, in.State.WorkflowID, cfg.Role)
 	if err != nil {
 		return ActionResult{}, fmt.Errorf("queue_run_for_each_participant list participants: %w", err)
 	}
 	reason := queueRunForEachParticipantReason(in)
-	for _, p := range all {
-		if p.Role != cfg.Role {
-			continue
-		}
+	for _, p := range seats {
 		req := QueueRunRequest{
 			AgentProfileID: p.AgentProfileID,
 			TaskID:         taskID,

apps/backend/internal/workflow/engine/queue_run_participant_slate_test.go

@@ -0,0 +1,185 @@
+package engine
+
+import (
+	"context"
+	"strings"
+	"testing"
+)
+
+// crossTaskParticipants emulates the real repository's two queries closely
+// enough to catch a cross-workflow leak: ListStepParticipants filters
+// `step_id = ? AND (task_id = ” OR task_id = ?)` (step-scoped, as the repo
+// actually enforces); ListTaskParticipants filters only `task_id = ?`,
+// ignoring step_id and workflow entirely — the unscoped, any-step query a
+// cross-task participant_role target must NOT use. Unlike scopedParticipants
+// (quorum_test.go), whose ListStepParticipants always returns nil for a
+// non-empty taskID, this fake answers a real step+task lookup so a leak from
+// the wrong query is actually observable.
+type crossTaskParticipants struct {
+	rows []ParticipantInfo
+}
+
+func (c crossTaskParticipants) ListStepParticipants(_ context.Context, stepID, taskID string) ([]ParticipantInfo, error) {
+	out := make([]ParticipantInfo, 0)
+	for _, p := range c.rows {
+		if p.StepID != stepID {
+			continue
+		}
+		if p.TaskID != "" && p.TaskID != taskID {
+			continue
+		}
+		out = append(out, p)
+	}
+	return out, nil
+}
+
+func (c crossTaskParticipants) ListTaskParticipants(_ context.Context, taskID string) ([]ParticipantInfo, error) {
+	out := make([]ParticipantInfo, 0)
+	for _, p := range c.rows {
+		if p.TaskID == taskID {
+			out = append(out, p)
+		}
+	}
+	return out, nil
+}
+
+// WO-30: a participant added while the task sat on an earlier step
+// (AddTaskParticipant stamps step_id at write time) must still be woken by
+// the fan-out when the task has since advanced — the quorum guard already
+// counts that participant via its any-step read (requiredSeatsForWorkflow),
+// so the fan-out reading a narrower, step-scoped population left a seat the
+// guard requires but the fan-out could never wake, and all_approve waited
+// forever. Uses scopedParticipants (quorum_test.go), which — unlike
+// fakeParticipants — correctly distinguishes a per-task row's real StepID
+// from a step-template row, exercising the case fakeParticipants cannot.
+func TestQueueRunForEachParticipantCallback_WakesParticipantAddedOnEarlierStep(t *testing.T) {
+	q := &fakeRunQueue{}
+	parts := scopedParticipants{
+		perTask: []ParticipantInfo{
+			{ID: "p1", StepID: "step-work", TaskID: "task-1", Role: "reviewer", AgentProfileID: "rev-A"},
+		},
+	}
+	cb := QueueRunForEachParticipantCallback{Adapter: q, Participants: parts}
+	in := ActionInput{
+		Trigger:     TriggerOnEnter,
+		State:       MachineState{TaskID: "task-1", WorkflowID: "wf-1"},
+		Step:        StepSpec{ID: "step-review"},
+		OperationID: "op-1",
+		Action: Action{
+			Kind: ActionQueueRunForEachParticipant,
+			QueueRunForEachParticipant: &QueueRunForEachParticipantAction{
+				Role: "reviewer",
+			},
+		},
+	}
+	if _, err := cb.Execute(context.Background(), in); err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if len(q.calls) != 1 {
+		t.Fatalf("expected 1 queue run call for a reviewer added on an earlier step, got %d", len(q.calls))
+	}
+	if q.calls[0].AgentProfileID != "rev-A" {
+		t.Fatalf("agent_profile_id = %q, want rev-A", q.calls[0].AgentProfileID)
+	}
+}
+
+// WO-30 regression guard: a reviewer present at BOTH the current step (as a
+// template row) and an earlier step (as a per-task row) must be woken
+// exactly once, not twice — the any-step gather can return the same
+// reviewer from both queries, and collapseByRoleAgent's per-task-wins-over-
+// template dedupe is what keeps that from double-queuing a run.
+func TestQueueRunForEachParticipantCallback_ReviewerAtBothStepsQueuesOnce(t *testing.T) {
+	q := &fakeRunQueue{}
+	parts := scopedParticipants{
+		perTask: []ParticipantInfo{
+			{ID: "p1", StepID: "step-work", TaskID: "task-1", Role: "reviewer", AgentProfileID: "rev-A"},
+		},
+		template: []ParticipantInfo{
+			{ID: "p2", StepID: "step-review", TaskID: "", Role: "reviewer", AgentProfileID: "rev-A"},
+		},
+	}
+	cb := QueueRunForEachParticipantCallback{Adapter: q, Participants: parts}
+	in := ActionInput{
+		Trigger:     TriggerOnEnter,
+		State:       MachineState{TaskID: "task-1", WorkflowID: "wf-1"},
+		Step:        StepSpec{ID: "step-review"},
+		OperationID: "op-1",
+		Action: Action{
+			Kind: ActionQueueRunForEachParticipant,
+			QueueRunForEachParticipant: &QueueRunForEachParticipantAction{
+				Role: "reviewer",
+			},
+		},
+	}
+	if _, err := cb.Execute(context.Background(), in); err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if len(q.calls) != 1 {
+		t.Fatalf("expected exactly 1 queue run call for a reviewer seated at both steps, got %d", len(q.calls))
+	}
+}
+
+// WO-30: the second call site (queue_run with target=participant_role:<role>)
+// has the identical defect — on the same arrangement it currently hard-errors
+// with "no participants with role", failing the whole action instead of
+// silently queuing zero.
+func TestQueueRunCallback_ParticipantRoleWakesParticipantAddedOnEarlierStep(t *testing.T) {
+	q := &fakeRunQueue{}
+	parts := scopedParticipants{
+		perTask: []ParticipantInfo{
+			{ID: "p1", StepID: "step-work", TaskID: "task-1", Role: "reviewer", AgentProfileID: "rev-A"},
+		},
+	}
+	cb := QueueRunCallback{Adapter: q, Participants: parts}
+	in := ActionInput{
+		Trigger: TriggerOnComment,
+		State:   MachineState{TaskID: "task-1", WorkflowID: "wf-1"},
+		Step:    StepSpec{ID: "step-review"},
+		Action: Action{
+			Kind: ActionQueueRun,
+			QueueRun: &QueueRunAction{
+				Target: "participant_role:reviewer",
+			},
+		},
+		OperationID: "op-1",
+	}
+	if _, err := cb.Execute(context.Background(), in); err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if len(q.calls) != 1 {
+		t.Fatalf("expected 1 queue run call for a reviewer added on an earlier step, got %d", len(q.calls))
+	}
+	if q.calls[0].AgentProfileID != "rev-A" {
+		t.Fatalf("agent_profile_id = %q, want rev-A", q.calls[0].AgentProfileID)
+	}
+}
+
+// WO-30 review round 1 regression: applying the any-step slate to a
+// CROSS-task participant_role target reads across workflows. A per-task row
+// stamped on a step the target task no longer uses (because it changed
+// workflow, or simply advanced) must not be woken by a cross-task fan-out —
+// unlike the same-task case, the engine has no target-task workflow id to
+// scope the any-step query with, so widening it here can only leak. The
+// cross-task path must stay step-scoped to the target task's CURRENT step,
+// exactly as it was before WO-30.
+func TestQueueRunCallback_ParticipantRoleCrossTaskDoesNotLeakStaleStepRow(t *testing.T) {
+	q := &fakeRunQueue{}
+	parts := crossTaskParticipants{
+		rows: []ParticipantInfo{
+			{ID: "p-stale", StepID: "step-old", TaskID: "target-task", Role: "reviewer", AgentProfileID: "rev-leaked"},
+		},
+	}
+	cb := QueueRunCallback{Adapter: q, Participants: parts, TaskSteps: fakeTaskSteps{id: "step-new"}}
+	in := newQueueRunInput("participant_role:reviewer", "target-task")
+
+	_, err := cb.Execute(context.Background(), in)
+	if err == nil {
+		t.Fatalf("expected error: cross-task fan-out must not find the stale per-task row from step-old")
+	}
+	if !strings.Contains(err.Error(), `no participants with role "reviewer"`) {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if len(q.calls) != 0 {
+		t.Fatalf("expected 0 queue run calls (stale row must not leak), got %d: %#v", len(q.calls), q.calls)
+	}
+}

apps/backend/internal/workflow/engine/queue_run_test.go

@@ -63,14 +63,18 @@ func (s *sequencePrimary) WorkflowStepIDForTask(_ context.Context, _ string) (st
 	return "step-1", nil
 }
 
-// fakeParticipants returns a static slice for any step.
+// fakeParticipants returns a static slice for any step. ListStepParticipants
+// and ListTaskParticipants both return the same f.list, matching (and
+// exercising) the union/dedupe path that gatherParticipantSlate feeds into
+// canonicalizeByTaskRoleAgent/collapseByRoleAgent.
 type fakeParticipants struct {
-	list          []ParticipantInfo
-	err           error
-	taskStepID    string
-	stepID        *string
-	taskID        *string
-	resolveTaskID *string
+	list              []ParticipantInfo
+	err               error
+	taskStepID        string
+	stepID            *string
+	taskID            *string
+	resolveTaskID     *string
+	taskParticipantID *string
 }
 
 func (f fakeParticipants) ListStepParticipants(_ context.Context, stepID, taskID string) ([]ParticipantInfo, error) {
@@ -90,7 +94,10 @@ func (f fakeParticipants) WorkflowStepIDForTask(_ context.Context, taskID string
 	return f.taskStepID, f.err
 }
 
-func (f fakeParticipants) ListTaskParticipants(_ context.Context, _ string) ([]ParticipantInfo, error) {
+func (f fakeParticipants) ListTaskParticipants(_ context.Context, taskID string) ([]ParticipantInfo, error) {
+	if f.taskParticipantID != nil {
+		*f.taskParticipantID = taskID
+	}
 	return f.list, f.err
 }
 
@@ -390,19 +397,28 @@ func TestQueueRunCallback_TargetParticipantRole(t *testing.T) {
 	}
 }
 
+// A cross-task participant_role target must stay step-scoped to the target
+// task's resolved current step (WO-30 review round 1: the any-step slate is
+// only safe for the same-task case, since only there does in.State.WorkflowID
+// scope the read to the task's actual current workflow — see
+// resolveParticipantRoleStepScoped). This asserts the single
+// ListStepParticipants(stepID, taskID) call the pre-WO-30 code made, and that
+// the any-step/any-workflow ListTaskParticipants query is never reached.
 func TestQueueRunCallback_ParticipantRoleUsesResolvedTargetStep(t *testing.T) {
 	q := &fakeRunQueue{}
 	var listedStepID string
-	var listedTaskID string
+	var listedStepTaskID string
+	var listedPerTaskID string
 	var stepResolverTaskID string
 	parts := fakeParticipants{
 		list: []ParticipantInfo{
 			{ID: "p-target", Role: "reviewer", AgentProfileID: "rev-target"},
 		},
-		taskStepID:    "target-step",
-		stepID:        &listedStepID,
-		taskID:        &listedTaskID,
-		resolveTaskID: &stepResolverTaskID,
+		taskStepID:        "target-step",
+		stepID:            &listedStepID,
+		taskID:            &listedStepTaskID,
+		resolveTaskID:     &stepResolverTaskID,
+		taskParticipantID: &listedPerTaskID,
 	}
 	cb := QueueRunCallback{Adapter: q, Participants: parts}
 	in := newQueueRunInput("participant_role:reviewer", "target-task")
@@ -420,8 +436,11 @@ func TestQueueRunCallback_ParticipantRoleUsesResolvedTargetStep(t *testing.T) {
 	if listedStepID != "target-step" {
 		t.Fatalf("participants listed for step %q, want target-step", listedStepID)
 	}
-	if listedTaskID != "target-task" {
-		t.Fatalf("participants listed for task %q, want target-task", listedTaskID)
+	if listedStepTaskID != "target-task" {
+		t.Fatalf("cross-task participant lookup must stay step-scoped to task_id=target-task, got %q", listedStepTaskID)
+	}
+	if listedPerTaskID != "" {
+		t.Fatalf("cross-task participant_role must not reach the any-step/any-workflow ListTaskParticipants query, got task_id %q", listedPerTaskID)
 	}
 	got := q.calls[0]
 	if got.WorkflowStepID != "target-step" {

apps/backend/internal/workflow/engine/quorum.go

@@ -194,24 +194,10 @@ func (e *Engine) requiredSeats(ctx context.Context, stepID, taskID, role string)
 }
 
 func (e *Engine) requiredSeatsForWorkflow(ctx context.Context, stepID, taskID, workflowID, role string) ([]ParticipantInfo, error) {
-	var perTask []ParticipantInfo
-	var err error
-	if scoped, ok := e.participants.(WorkflowScopedParticipantStore); ok && workflowID != "" {
-		perTask, err = scoped.ListTaskParticipantsForWorkflow(ctx, taskID, workflowID)
-	} else {
-		perTask, err = e.participants.ListTaskParticipants(ctx, taskID)
-	}
+	gathered, err := gatherParticipantSlate(ctx, e.participants, stepID, taskID, workflowID)
 	if err != nil {
-		return nil, fmt.Errorf("list task participants for quorum: %w", err)
+		return nil, err
 	}
-	template, err := e.participants.ListStepParticipants(ctx, stepID, "")
-	if err != nil {
-		return nil, fmt.Errorf("list step participants for quorum: %w", err)
-	}
-
-	gathered := make([]ParticipantInfo, 0, len(perTask)+len(template))
-	gathered = append(gathered, perTask...)
-	gathered = append(gathered, template...)
 
 	filtered := make([]ParticipantInfo, 0, len(gathered))
 	for _, p := range gathered {
@@ -225,6 +211,35 @@ func (e *Engine) requiredSeatsForWorkflow(ctx context.Context, stepID, taskID, w
 	return collapseByRoleAgent(canonical), nil
 }
 
+// gatherParticipantSlate implements AC-50 step 1, shared by the quorum guard
+// and every fan-out/resolution call site that must see the same participant
+// population the guard counts: per-task rows for taskID at ANY step
+// (workflow-scoped when the store supports it and workflowID is known)
+// unioned with template rows (task_id="") at the evaluating step only. It
+// does not filter by role/decision_required or dedupe — callers apply their
+// own filter and canonicalizeByTaskRoleAgent/collapseByRoleAgent afterward.
+func gatherParticipantSlate(ctx context.Context, store ParticipantStore, stepID, taskID, workflowID string) ([]ParticipantInfo, error) {
+	var perTask []ParticipantInfo
+	var err error
+	if scoped, ok := store.(WorkflowScopedParticipantStore); ok && workflowID != "" {
+		perTask, err = scoped.ListTaskParticipantsForWorkflow(ctx, taskID, workflowID)
+	} else {
+		perTask, err = store.ListTaskParticipants(ctx, taskID)
+	}
+	if err != nil {
+		return nil, fmt.Errorf("list task participants: %w", err)
+	}
+	template, err := store.ListStepParticipants(ctx, stepID, "")
+	if err != nil {
+		return nil, fmt.Errorf("list step participants: %w", err)
+	}
+
+	gathered := make([]ParticipantInfo, 0, len(perTask)+len(template))
+	gathered = append(gathered, perTask...)
+	gathered = append(gathered, template...)
+	return gathered, nil
+}
+
 // seatCanonKey identifies a row for AC-44 canonicalization: (task_id, role,
 // agent_profile_id), except a row with no agent_profile_id has no identity
 // to canonicalize on and is kept standalone, keyed by its own row id.