diff --git a/apps/backend/internal/workflow/engine/phase2_callbacks.go b/apps/backend/internal/workflow/engine/phase2_callbacks.go index 634a6a9ecf..af0fac11b2 100644 --- a/apps/backend/internal/workflow/engine/phase2_callbacks.go +++ b/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, diff --git a/apps/backend/internal/workflow/engine/queue_run_participant_slate_test.go b/apps/backend/internal/workflow/engine/queue_run_participant_slate_test.go new file mode 100644 index 0000000000..14dcccfb23 --- /dev/null +++ b/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:) +// 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) + } +} diff --git a/apps/backend/internal/workflow/engine/queue_run_test.go b/apps/backend/internal/workflow/engine/queue_run_test.go index 25733694e3..15dd5bf807 100644 --- a/apps/backend/internal/workflow/engine/queue_run_test.go +++ b/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" { diff --git a/apps/backend/internal/workflow/engine/quorum.go b/apps/backend/internal/workflow/engine/quorum.go index dac07ee37c..ddffe79932 100644 --- a/apps/backend/internal/workflow/engine/quorum.go +++ b/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.