package run

import (
	"context"
	"fmt"
	"strings"
	"time"

	"github.com/tidbcloud/AgentTeam/internal/core/identifiers"
	"github.com/tidbcloud/AgentTeam/internal/coredb"
	"github.com/tidbcloud/AgentTeam/internal/store/model"
	"github.com/tidbcloud/AgentTeam/internal/store/storeerr"
	"github.com/tidbcloud/AgentTeam/internal/workflow/plan"
)

// candidateWaiterScanItem is one durable FIFO candidate-pool waiter selected for
// a promotion pass.
type candidateWaiterScanItem struct {
	workspace, run, activation, effect, node, poolKey string
	candidates                                        []string
	waiterSeq                                         int64
	globalSeq                                         int64
	selectOneStrategy                                 plan.SelectOneStrategy
}

// PromoteCallbackResponderWaitersParams carries the promoting worker's identity
// and dispatch-lease TTL so promoted effects are claimed by the caller, not a
// synthetic worker.
type PromoteCallbackResponderWaitersParams struct {
	// WorkerID is the promoting worker that will deliver the promoted effects.
	WorkerID string
	// LeaseTTL is the dispatch-lease TTL applied to promoted effects.
	LeaseTTL time.Duration
	// Now is the promotion timestamp; zero defaults to time.Now().
	Now time.Time
	// ScanLimit bounds how many distinct candidates one promotion pass examines
	// (the "scan budget"). Candidates are paged through in stable
	// candidate_agent_id order with a durable per-workspace candidate cursor, so
	// a candidate beyond the budget in one pass is examined in a later pass —
	// bounded work per reconcile and durable progress across restart. Zero
	// defaults to the caller's limit.
	ScanLimit int
	// Randomizer is the injectable unbiased source used when a promoted waiter
	// declared the "random" selectOne strategy. It picks uniformly at random
	// among the waiter's currently-available deduplicated candidates; nil
	// defaults to a crypto-backed source so production is never predictable. An
	// injected Randomizer that returns a bounded error proves the crypto-failure
	// path fails closed (the member is left waiting, nothing is claimed).
	Randomizer Randomizer
}

// candidateWaiterScanPageSize bounds the keyset candidate page read per query
// within the scan budget. It is deliberately small so one pass can cross many
// candidates without unbounded work.
const candidateWaiterScanPageSize = 64

// PromoteCallbackResponderWaiters wakes waiting_for_responder waiters whose
// candidate pool now has a free candidate, bounded per pass. For each promoted
// effect it atomically selects the highest-priority free candidate, claims its
// execution lease, freezes it, transitions the effect from
// waiting_for_responder to dispatching with a fresh dispatch lease, and marks
// the waiter selected — the caller must deliver the returned effects.
//
// The scan is per-candidate FIFO, NOT a waiter scan and NOT a waiter cursor:
//
//   - For each distinct candidate C that has at least one live waiting member,
//     the OLDEST member (min global_seq, the workspace arrival order) is
//     promoted. The membership index (workflow_responder_candidate_members) is
//     keyed by (workspace, candidate, effect) and ordered by the waiter's
//     global_seq, so a freed candidate always finds its oldest live waiter — a
//     newer waiter can never jump an older one for the same candidate. An older
//     waiter that is currently unsatisfiable stays the oldest member and is
//     retried, so it retains global-arrival priority whenever it becomes
//     satisfiable.
//   - Candidates with independent free capacity are processed independently: a
//     long unsatisfiable prefix on one candidate (its capacity still busy) does
//     not block a later satisfiable waiter on another candidate.
//   - Candidates are paged through in stable candidate_agent_id order within
//     the ScanLimit budget, using a durable per-workspace candidate cursor
//     (last_scanned_candidate_agent_id). The cursor only selects WHICH
//     candidates are examined this pass; within a candidate the oldest member is
//     always chosen, so contention order is preserved. No waiter cursor is used,
//     so an older waiter that becomes satisfiable later is never skipped. The
//     candidate cursor survives restart/cancel/new waiters because it is only a
//     scan position, never a promotion decision.
//   - Membership rows are maintained by syncSelectOneMembership on waiter
//     insert/refresh and deleted by deleteSelectOneMembership on terminalize, so
//     a promoted, canceled, or completed waiter disappears from its candidates'
//     queues.
//
// It is a separate bounded pass (like ReconcileAgentExecutionLeases) run by the
// worker; it never pokes another Run, so lock order stays
// placement -> Run -> Effect -> waiter -> candidate execution lease -> queue.
func (store *Store) PromoteCallbackResponderWaiters(ctx context.Context, workspaceGroupID string, params PromoteCallbackResponderWaitersParams, limit int) ([]SelectCallbackResponderCapacityResult, error) {
	if err := store.Validate(); err != nil {
		return nil, err
	}
	params.WorkerID = strings.TrimSpace(params.WorkerID)
	if params.WorkerID == "" {
		return nil, storeerr.BadRequest("promoting workerId is required")
	}
	if params.LeaseTTL <= 0 {
		return nil, storeerr.BadRequest("positive dispatch leaseTTL is required")
	}
	if params.Now.IsZero() {
		params.Now = time.Now()
	}
	params.Now = utc(params.Now)
	bounded, err := boundedLimit(limit)
	if err != nil {
		return nil, err
	}
	scanBudget := params.ScanLimit
	if scanBudget <= 0 {
		scanBudget = bounded
	}
	scanBudget, err = boundedLimit(scanBudget)
	if err != nil {
		return nil, err
	}
	if workspaceGroupID == "" {
		workspaceGroupID = store.workspaceGroupID
	}

	// Workspaces in the group that currently have waiting waiters. Each keeps
	// its own durable candidate cursor, so one workspace's unsatisfiable prefix
	// cannot starve another workspace's satisfiable waiters.
	workspaceRows, err := store.db.Query(ctx, `
SELECT DISTINCT w.workspace_id
FROM workflow_responder_candidate_waiters w
JOIN workflow_run_instances run
  ON run.workspace_id = w.workspace_id AND run.id = w.run_id
JOIN workspace_placements placement ON placement.workspace_id = w.workspace_id
WHERE w.status = 'waiting'
  AND placement.workspace_group_id = COALESCE(NULLIF($1,''), placement.workspace_group_id)
  AND placement.state IN ('active','draining','moving')
  AND run.status IN ('running','waiting')
ORDER BY w.workspace_id ASC`, workspaceGroupID)
	if err != nil {
		return nil, fmt.Errorf("scan workflow candidate-pool workspaces for promotion: %w", err)
	}
	var workspaceIDs []string
	for workspaceRows.Next() {
		var id string
		if err := workspaceRows.Scan(&id); err != nil {
			workspaceRows.Close()
			return nil, fmt.Errorf("scan workflow candidate-pool workspace: %w", err)
		}
		workspaceIDs = append(workspaceIDs, id)
	}
	workspaceRows.Close()
	if err := workspaceRows.Err(); err != nil {
		return nil, fmt.Errorf("iterate workflow candidate-pool workspaces: %w", err)
	}

	var promoted []SelectCallbackResponderCapacityResult
	for _, workspaceID := range workspaceIDs {
		if len(promoted) >= bounded {
			break
		}
		oneWorkspace, err := store.promoteWorkspaceWaiters(ctx, workspaceID, params, scanBudget, bounded-len(promoted))
		if err != nil {
			return promoted, err
		}
		promoted = append(promoted, oneWorkspace...)
	}
	return promoted, nil
}

// promoteWorkspaceWaiters scans ONE workspace's distinct candidates that have
// waiting members, in stable candidate_agent_id order starting from its durable
// candidate cursor, within a bounded scan budget. For each examined candidate
// it promotes that candidate's OLDEST live waiting member (per-candidate FIFO).
// The candidate cursor advances past every examined candidate (whether it
// promoted or not) and wraps to the start when the end is reached, so repeated
// calls page through all candidates with waiting members without ever reordering
// waiters within a candidate.
func (store *Store) promoteWorkspaceWaiters(
	ctx context.Context,
	workspaceID string,
	params PromoteCallbackResponderWaitersParams,
	scanBudget int,
	resultCap int,
) ([]SelectCallbackResponderCapacityResult, error) {
	cursor, err := store.loadCandidateScanCursor(ctx, workspaceID)
	if err != nil {
		return nil, err
	}
	var promoted []SelectCallbackResponderCapacityResult
	examined := 0
	wrapped := false
	for examined < scanBudget && len(promoted) < resultCap {
		page := candidateWaiterScanPageSize
		if remaining := scanBudget - examined; remaining < page {
			page = remaining
		}
		candidates, err := store.scanCandidatePage(ctx, workspaceID, cursor, page)
		if err != nil {
			return promoted, err
		}
		if len(candidates) == 0 {
			// No candidate newer than the cursor: the pass swept the whole set.
			// Wrap the cursor so the next call re-examines candidates from the
			// start (their oldest members may have become satisfiable since).
			if wrapped {
				break
			}
			cursor = ""
			wrapped = true
			continue
		}
		for _, candidate := range candidates {
			if examined >= scanBudget || len(promoted) >= resultCap {
				break
			}
			examined++
			cursor = candidate
			one, err := store.promoteOldestMemberForCandidate(ctx, workspaceID, candidate, params)
			if err != nil {
				return promoted, err
			}
			if one != nil {
				promoted = append(promoted, *one)
			}
		}
	}
	if err := store.saveCandidateScanCursor(ctx, workspaceID, cursor); err != nil {
		return promoted, err
	}
	return promoted, nil
}

// scanCandidatePage reads up to limit distinct candidates in one workspace that
// have at least one live waiting member, in candidate_agent_id keyset order
// strictly after the given cursor (no OFFSET). The join with the live waiter
// rows keeps the page bounded to candidates that can actually promote.
func (store *Store) scanCandidatePage(ctx context.Context, workspaceID, afterCandidate string, limit int) ([]string, error) {
	rows, err := store.db.Query(ctx, `
SELECT DISTINCT m.candidate_agent_id
FROM workflow_responder_candidate_members m
JOIN workflow_responder_candidate_waiters w
  ON w.workspace_id = m.workspace_id AND w.effect_id = m.effect_id
JOIN workflow_run_instances run
  ON run.workspace_id = w.workspace_id AND run.id = w.run_id
JOIN workspace_placements placement ON placement.workspace_id = w.workspace_id
WHERE m.workspace_id = $1 AND m.candidate_agent_id > $2
  AND w.status = 'waiting'
  AND placement.state IN ('active','draining','moving')
  AND run.status IN ('running','waiting')
ORDER BY m.candidate_agent_id ASC
LIMIT $3`, workspaceID, afterCandidate, limit)
	if err != nil {
		return nil, fmt.Errorf("scan workflow candidate-pool candidates for promotion: %w", err)
	}
	defer rows.Close()
	candidates := make([]string, 0, limit)
	for rows.Next() {
		var candidate string
		if err := rows.Scan(&candidate); err != nil {
			return nil, fmt.Errorf("scan workflow candidate-pool candidate: %w", err)
		}
		candidates = append(candidates, candidate)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("iterate workflow candidate-pool candidates: %w", err)
	}
	return candidates, nil
}

// promoteOldestMemberForCandidate promotes, for one candidate, the OLDEST live
// waiting member (min global_seq, workspace arrival order) for which that
// candidate is the member's SELECTED candidate — i.e. for an ordered member all
// higher-priority candidates in the member's ordered pool are busy, so the freed
// candidate C is the first free candidate that member would take; a "random"
// member draws uniformly among its currently-available candidates (honoring the
// persisted random strategy). It is inherently FIFO per candidate: a freed
// candidate is always offered to its oldest waiter that actually wants it, so a
// newer waiter can never jump an older one, and a member that prefers a
// higher-priority free candidate B is left for B's own promotion (never stealing
// B from an older waiter). promoteMemberSelectingCandidate /
// promoteMemberSelectingRandomCandidate atomically claim under the
// placement->Run->effect->waiter->lease lock order.
func (store *Store) promoteOldestMemberForCandidate(
	ctx context.Context,
	workspaceID, candidate string,
	params PromoteCallbackResponderWaitersParams,
) (*SelectCallbackResponderCapacityResult, error) {
	items, err := store.waitingMembersForCandidate(ctx, store.db, workspaceID, candidate, candidateWaiterScanPageSize)
	if err != nil {
		return nil, err
	}
	for _, item := range items {
		var result *SelectCallbackResponderCapacityResult
		var targetBusy bool
		if item.selectOneStrategy.Normalized() == plan.SelectOneStrategyRandom {
			result, targetBusy, err = store.promoteMemberSelectingRandomCandidate(ctx, params, item)
		} else {
			result, targetBusy, err = store.promoteMemberSelectingCandidate(ctx, params, item, candidate)
		}
		if err != nil {
			return nil, err
		}
		if targetBusy {
			// The candidate has no free capacity this pass; no later member can
			// select it either.
			return nil, nil
		}
		if result != nil {
			return result, nil
		}
		// This member prefers a higher-priority free candidate; its own candidate
		// promotion will claim it. Try the next-oldest member of candidate.
	}
	return nil, nil
}

// waitingMembersForCandidate returns up to limit live waiting members of one
// candidate in workspace arrival order (min global_seq via the membership
// index, tie-break effect_id), bounded for the per-candidate scan budget. The
// queryer may be the store DB or the caller's transaction (for tx-scoped
// capacity arbitration), so the caller controls snapshot visibility.
func (store *Store) waitingMembersForCandidate(ctx context.Context, q coredb.Queryer, workspaceID, candidate string, limit int) ([]candidateWaiterScanItem, error) {
	rows, err := q.Query(ctx, `
SELECT w.workspace_id, w.run_id, w.activation_id, w.effect_id, w.node_id,
       w.candidate_pool_key, w.candidate_pool_json, w.select_one_strategy, w.waiter_seq, w.global_seq
FROM workflow_responder_candidate_members m
JOIN workflow_responder_candidate_waiters w
  ON w.workspace_id = m.workspace_id AND w.effect_id = m.effect_id
JOIN workflow_run_instances run
  ON run.workspace_id = w.workspace_id AND run.id = w.run_id
JOIN workspace_placements placement ON placement.workspace_id = w.workspace_id
WHERE m.workspace_id = $1 AND m.candidate_agent_id = $2
  AND w.status = 'waiting'
  AND placement.state IN ('active','draining','moving')
  AND run.status IN ('running','waiting')
ORDER BY m.global_seq ASC, m.effect_id ASC
LIMIT $3`, workspaceID, candidate, limit)
	if err != nil {
		return nil, fmt.Errorf("scan workflow candidate-pool members for promotion: %w", err)
	}
	defer rows.Close()
	items := make([]candidateWaiterScanItem, 0, limit)
	for rows.Next() {
		var item candidateWaiterScanItem
		var poolJSON []byte
		var storedStrategy string
		if err := rows.Scan(&item.workspace, &item.run, &item.activation, &item.effect, &item.node,
			&item.poolKey, &poolJSON, &storedStrategy, &item.waiterSeq, &item.globalSeq); err != nil {
			return nil, fmt.Errorf("scan workflow candidate-pool member: %w", err)
		}
		if err := model.UnmarshalJSONObject("workflow responder candidate pool", poolJSON, &item.candidates); err != nil {
			return nil, fmt.Errorf("decode workflow candidate-pool member: %w", err)
		}
		parsedStrategy, parseErr := parseStoredSelectOneStrategy(storedStrategy)
		if parseErr != nil {
			return nil, parseErr
		}
		item.selectOneStrategy = parsedStrategy
		items = append(items, item)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("iterate workflow candidate-pool members: %w", err)
	}
	return items, nil
}

// memberSelectsCandidateLocked reports whether a live waiting member would
// CURRENTLY select the given target candidate. For an ordered member it iterates
// the member's ordered pool: every higher-priority candidate is busy and the
// target has free capacity (or is already owned by this same member on retry).
// A "random" member instead draws uniformly among all currently-available pool
// candidates, so it reserves every available one. It is the exact eligibility
// predicate the promotion path uses in promoteMemberSelectingCandidate, shared
// so the immediate dispatch path and the promotion path agree on which older
// waiter is eligible for a freed candidate. The caller must hold the placement
// lock and the member's candidate leases are locked FOR UPDATE in pool-priority
// order (consistent with claim/cancel/reconcile). The effectID is the member's
// own effect.
func memberSelectsCandidateLocked(
	ctx context.Context,
	tx coredb.Tx,
	store *Store,
	workspaceID string,
	item candidateWaiterScanItem,
	target string,
) (bool, error) {
	random := item.selectOneStrategy.Normalized() == plan.SelectOneStrategyRandom
	for _, candidate := range item.candidates {
		lease, leaseFound, err := store.lockAgentExecutionLease(ctx, tx, workspaceID, candidate)
		if err != nil {
			return false, err
		}
		free, err := candidateFreeCapacityLocked(ctx, tx, workspaceID, candidate, item.effect, lease, leaseFound)
		if err != nil {
			return false, err
		}
		if candidate == target {
			return free, nil
		}
		if free && !random {
			// An ordered member selects its highest-priority free candidate, not
			// the target; a random member would include every free candidate in
			// its draw, so it reserves each one.
			return false, nil
		}
	}
	// No candidate in the member's pool has free capacity (all busy).
	return false, nil
}

// candidateReservedByOlderWaiter reports whether any OLDER live waiting member
// of the candidate is currently eligible to claim that candidate under
// ordered-pool preference. It is the FIFO arbitration guard for the immediate
// dispatch path: a new-arrival effect may claim candidate C only if no older
// live waiting member would currently select C. The caller must hold the
// placement lock; the member scan is tx-scoped so the capacity state is read in
// the same atomic snapshot as the claim decision (no check-then-act race).
// newcomerEffectID excludes the calling effect (which has no waiter row yet on
// the immediate path, but is passed for retry/recovery safety).
func (store *Store) candidateReservedByOlderWaiter(
	ctx context.Context,
	tx coredb.Tx,
	workspaceID, candidate, newcomerEffectID string,
) (bool, error) {
	members, err := store.waitingMembersForCandidate(ctx, tx, workspaceID, candidate, candidateWaiterScanPageSize)
	if err != nil {
		return false, err
	}
	for _, member := range members {
		if member.effect == newcomerEffectID {
			continue
		}
		selected, err := memberSelectsCandidateLocked(ctx, tx, store, workspaceID, member, candidate)
		if err != nil {
			return false, err
		}
		if selected {
			return true, nil
		}
	}
	return false, nil
}

// candidateReservedByOlderWaiterBefore reports whether any live waiting member
// OLDER than beforeGlobalSeq (strictly) of the candidate would currently select
// it. It is the promotion-path counterpart of candidateReservedByOlderWaiter: on
// the immediate path a newcomer has no waiter row yet, so every existing member
// is older; during promotion the member being offered ALREADY has a waiter row,
// so only members that arrived strictly earlier may reserve a candidate it might
// draw. Members are scanned in global_seq ascending order, so the bound stops at
// the promoted member itself and never considers newer members as reserving.
func (store *Store) candidateReservedByOlderWaiterBefore(
	ctx context.Context,
	tx coredb.Tx,
	workspaceID, candidate string,
	newcomerEffectID string,
	beforeGlobalSeq int64,
) (bool, error) {
	members, err := store.waitingMembersForCandidate(ctx, tx, workspaceID, candidate, candidateWaiterScanPageSize)
	if err != nil {
		return false, err
	}
	for _, member := range members {
		if beforeGlobalSeq > 0 && member.globalSeq >= beforeGlobalSeq {
			return false, nil
		}
		if member.effect == newcomerEffectID {
			continue
		}
		selected, err := memberSelectsCandidateLocked(ctx, tx, store, workspaceID, member, candidate)
		if err != nil {
			return false, err
		}
		if selected {
			return true, nil
		}
	}
	return false, nil
}

// promoteMemberSelectingCandidate promotes an ORDERED member for the exact
// target candidate when the target is the member's selected candidate (every
// higher-priority pool candidate is busy, or owned by this same effect on
// retry), and the target is free. It returns targetBusy=true when the target has
// no free capacity this pass, so the caller stops scanning the target's members.
// When the member prefers a higher-priority FREE candidate, it returns
// (nil,false) so the caller leaves that member to the higher-priority
// candidate's own promotion — the member is NOT stolen from an older waiter of
// that candidate.
func (store *Store) promoteMemberSelectingCandidate(
	ctx context.Context,
	params PromoteCallbackResponderWaitersParams,
	item candidateWaiterScanItem,
	target string,
) (*SelectCallbackResponderCapacityResult, bool, error) {
	var promoted *SelectCallbackResponderCapacityResult
	targetBusy := false
	err := coredb.WithTx(ctx, store.db, func(tx coredb.Tx) error {
		if _, err := store.lockCandidatePlacement(ctx, tx, item.workspace); err != nil {
			return err
		}
		// Lock the owning Run, then the effect, then the waiter row, then the
		// candidate leases — the same order as claim/cancel/reconcile.
		run, err := scanRun(tx.QueryRow(ctx, runSelectSQL()+`
WHERE r.workspace_id=$1 AND r.id=$2 FOR UPDATE`, item.workspace, item.run))
		switch {
		case coredb.IsNotFound(err):
			return nil
		case err != nil:
			return fmt.Errorf("lock workflow Run for candidate-pool promotion: %w", err)
		case run.Status != RunStatusRunning && run.Status != RunStatusWaiting:
			return nil
		}
		effect, err := scanEffect(tx.QueryRow(ctx, effectSelectSQL()+`
WHERE workspace_id=$1 AND run_id=$2 AND id=$3 FOR UPDATE`, item.workspace, item.run, item.effect))
		if err != nil {
			return mapReadError("lock workflow Run effect for candidate-pool promotion", err)
		}
		// Revalidate the effect is still durably waiting for a responder and the
		// waiter is still live.
		if effect.Status != EffectStatusWaitingForResponder {
			return nil
		}
		var waiterStatus string
		if err := tx.QueryRow(ctx, `
SELECT status
FROM workflow_responder_candidate_waiters
WHERE workspace_id=$1 AND effect_id=$2 FOR UPDATE`, item.workspace, item.effect).Scan(&waiterStatus); err != nil {
			return mapReadError("lock workflow candidate-pool waiter for promotion", err)
		}
		if waiterStatus != ResponderWaiterStatusWaiting {
			return nil
		}
		// Try candidates in priority order. Only claim the target when it is the
		// member's selected candidate: every higher-priority candidate is busy
		// (or owned by this same effect on retry), and the target is free.
		selected := false
		for _, candidate := range item.candidates {
			lease, leaseFound, err := store.lockAgentExecutionLease(ctx, tx, item.workspace, candidate)
			if err != nil {
				return err
			}
			free, err := candidateFreeCapacityLocked(ctx, tx, item.workspace, candidate, item.effect, lease, leaseFound)
			if err != nil {
				return err
			}
			if candidate == target {
				if free {
					selected = true
				} else {
					targetBusy = true
				}
				break
			}
			// A higher-priority candidate that is free means this member does not
			// select the target; leave it to that candidate's own promotion.
			if free {
				break
			}
		}
		if !selected {
			return nil
		}
		result := SelectCallbackResponderCapacityResult{}
		if err := promoteSelectOneCandidate(ctx, tx, store, &result, effect, params, item, target, run); err != nil {
			return err
		}
		if _, err := tx.Exec(ctx, `
UPDATE workflow_responder_candidate_waiters
SET status='selected', updated_at=$3
WHERE workspace_id=$1 AND effect_id=$2 AND status='waiting'`, item.workspace, item.effect, params.Now); err != nil {
			return fmt.Errorf("mark workflow candidate-pool waiter selected: %w", err)
		}
		if err := deleteSelectOneMembership(ctx, tx, item.workspace, item.effect); err != nil {
			return err
		}
		copy := result
		promoted = &copy
		return nil
	})
	return promoted, targetBusy, err
}

// promoteMemberSelectingRandomCandidate promotes a "random"-strategy member
// (the waiter row persisted the random strategy when it parked). Under the same
// placement->Run->effect->waiter->lease lock order it gathers every candidate in
// the member's pool that currently has free capacity and is NOT reserved by an
// older live waiter, then draws one uniformly at random with the injectable
// randomizer. It is FIFO because the caller offers the candidate's oldest member
// first and the draw is bounded to non-reserved candidates, so a newer waiter can
// never jump an older one for any candidate. When the member has no available
// candidate (its pool is fully busy or reserved), it returns targetBusy=true so
// the caller stops scanning the target's members (the target itself must be
// busy, else it would be available).
func (store *Store) promoteMemberSelectingRandomCandidate(
	ctx context.Context,
	params PromoteCallbackResponderWaitersParams,
	item candidateWaiterScanItem,
) (*SelectCallbackResponderCapacityResult, bool, error) {
	var promoted *SelectCallbackResponderCapacityResult
	noAvailable := false
	err := coredb.WithTx(ctx, store.db, func(tx coredb.Tx) error {
		if _, err := store.lockCandidatePlacement(ctx, tx, item.workspace); err != nil {
			return err
		}
		run, err := scanRun(tx.QueryRow(ctx, runSelectSQL()+`
WHERE r.workspace_id=$1 AND r.id=$2 FOR UPDATE`, item.workspace, item.run))
		switch {
		case coredb.IsNotFound(err):
			return nil
		case err != nil:
			return fmt.Errorf("lock workflow Run for candidate-pool random promotion: %w", err)
		case run.Status != RunStatusRunning && run.Status != RunStatusWaiting:
			return nil
		}
		effect, err := scanEffect(tx.QueryRow(ctx, effectSelectSQL()+`
WHERE workspace_id=$1 AND run_id=$2 AND id=$3 FOR UPDATE`, item.workspace, item.run, item.effect))
		if err != nil {
			return mapReadError("lock workflow Run effect for candidate-pool random promotion", err)
		}
		if effect.Status != EffectStatusWaitingForResponder {
			return nil
		}
		var waiterStatus string
		if err := tx.QueryRow(ctx, `
SELECT status
FROM workflow_responder_candidate_waiters
WHERE workspace_id=$1 AND effect_id=$2 FOR UPDATE`, item.workspace, item.effect).Scan(&waiterStatus); err != nil {
			return mapReadError("lock workflow candidate-pool waiter for random promotion", err)
		}
		if waiterStatus != ResponderWaiterStatusWaiting {
			return nil
		}
		// Gather currently-available candidates, excluding any reserved by an
		// older live waiter (FIFO arbitration, same atomic placement snapshot).
		available := make([]string, 0, len(item.candidates))
		for _, candidate := range item.candidates {
			lease, leaseFound, err := store.lockAgentExecutionLease(ctx, tx, item.workspace, candidate)
			if err != nil {
				return err
			}
			free, err := candidateFreeCapacityLocked(ctx, tx, item.workspace, candidate, item.effect, lease, leaseFound)
			if err != nil {
				return err
			}
			if !free {
				continue
			}
			// Only a member OLDER than the promoted member may reserve this
			// candidate for itself; a newer member must not block the draw.
			reserved, err := store.candidateReservedByOlderWaiterBefore(ctx, tx, item.workspace, candidate, item.effect, item.globalSeq)
			if err != nil {
				return err
			}
			if reserved {
				continue
			}
			available = append(available, candidate)
		}
		if len(available) == 0 {
			noAvailable = true
			return nil
		}
		draw, err := selectOneRandomizer(params.Randomizer)(len(available))
		if err != nil {
			return err
		}
		chosen := available[draw]
		result := SelectCallbackResponderCapacityResult{}
		if err := promoteSelectOneCandidate(ctx, tx, store, &result, effect, params, item, chosen, run); err != nil {
			return err
		}
		if _, err := tx.Exec(ctx, `
UPDATE workflow_responder_candidate_waiters
SET status='selected', updated_at=$3
WHERE workspace_id=$1 AND effect_id=$2 AND status='waiting'`, item.workspace, item.effect, params.Now); err != nil {
			return fmt.Errorf("mark workflow candidate-pool waiter selected: %w", err)
		}
		if err := deleteSelectOneMembership(ctx, tx, item.workspace, item.effect); err != nil {
			return err
		}
		copy := result
		promoted = &copy
		return nil
	})
	return promoted, noAvailable, err
}

// loadCandidateScanCursor returns the durable per-workspace candidate scan
// cursor (the last candidate_agent_id examined), or "" when never scanned.
func (store *Store) loadCandidateScanCursor(ctx context.Context, workspaceID string) (string, error) {
	workspaceID = strings.TrimSpace(workspaceID)
	if workspaceID == "" {
		return "", storeerr.BadRequest("workspaceId is required")
	}
	var last string
	err := store.db.QueryRow(ctx, `
SELECT last_scanned_candidate_agent_id
FROM workflow_responder_candidate_scan_state
WHERE workspace_id=$1`, workspaceID).Scan(&last)
	if coredb.IsNotFound(err) {
		return "", nil
	}
	if err != nil {
		return "", fmt.Errorf("load workflow candidate-pool scan cursor: %w", err)
	}
	return last, nil
}

// LoadCallbackResponderScanCursor returns the durable per-workspace candidate
// scan cursor (the last candidate_agent_id examined, or "" when never scanned).
// It is an observability read for tests and diagnostics.
func (store *Store) LoadCallbackResponderScanCursor(ctx context.Context, workspaceID string) (string, error) {
	if err := store.Validate(); err != nil {
		return "", err
	}
	return store.loadCandidateScanCursor(ctx, workspaceID)
}

// saveCandidateScanCursor persists the per-workspace candidate scan cursor. The
// cursor is only a scan position, never a promotion decision: an older value
// written by a concurrent promoter merely re-scans a bounded candidate prefix
// next pass, always choosing each candidate's oldest member.
func (store *Store) saveCandidateScanCursor(ctx context.Context, workspaceID, lastCandidate string) error {
	workspaceID = strings.TrimSpace(workspaceID)
	if workspaceID == "" {
		return storeerr.BadRequest("workspaceId is required")
	}
	_, err := store.db.Exec(ctx, `
INSERT INTO workflow_responder_candidate_scan_state (workspace_id, last_scanned_candidate_agent_id, updated_at)
VALUES ($1,$2,$3)
ON DUPLICATE KEY UPDATE last_scanned_candidate_agent_id=VALUES(last_scanned_candidate_agent_id), updated_at=VALUES(updated_at)`,
		workspaceID, lastCandidate, utc(time.Now()))
	if err != nil {
		return fmt.Errorf("save workflow candidate-pool scan cursor: %w", err)
	}
	return nil
}

// promoteSelectOneCandidate freezes the chosen candidate and transitions the
// waiting_for_responder effect to dispatching with a fresh dispatch lease owned
// by the promoting worker, acquiring the candidate's execution lease and its
// FIFO position.
func promoteSelectOneCandidate(
	ctx context.Context,
	tx coredb.Tx,
	store *Store,
	result *SelectCallbackResponderCapacityResult,
	effect Effect,
	params PromoteCallbackResponderWaitersParams,
	item candidateWaiterScanItem,
	candidate string,
	run Run,
) error {
	now := params.Now
	leaseToken, err := identifiers.NewEntityID("wfl")
	if err != nil {
		return fmt.Errorf("generate workflow Run effect lease: %w", err)
	}
	claim := ClaimCallbackEffectCapacityParams{
		WorkspaceID: item.workspace, RunID: item.run, ActivationID: item.activation,
		EffectID: item.effect, WorkerID: params.WorkerID, Now: now, LeaseTTL: params.LeaseTTL,
		LeaseToken: leaseToken, ResponderAgentID: candidate, NodeID: item.node,
	}
	if err := upsertHeldCallbackLease(ctx, tx, claim, now); err != nil {
		return err
	}
	// Effect row: status dispatching + fresh dispatch lease + frozen responder.
	expires := utc(now.Add(claim.LeaseTTL))
	if _, err := tx.Exec(ctx, `
UPDATE workflow_run_effects
SET status='dispatching', callback_responder_agent_id=$4,
    lease_owner=$5, lease_token=$6, lease_expires_at=$7,
    queued_at=NULL, updated_at=$8
WHERE workspace_id=$1 AND run_id=$2 AND id=$3 AND status='waiting_for_responder'`,
		item.workspace, item.run, item.effect, candidate, claim.WorkerID, leaseToken, expires, now); err != nil {
		return fmt.Errorf("promote workflow Run callback to dispatching: %w", err)
	}
	if _, err := ensureCallbackQueueItem(ctx, tx, claim, callbackQueueStatusDispatching, now); err != nil {
		return err
	}
	if err := store.appendResponderCapacityEvent(ctx, tx, run, effect, item.node, EventTypeEffectResponderPromoted, now); err != nil {
		return err
	}
	effect.Status = EffectStatusDispatching
	effect.CallbackResponderAgentID = candidate
	effect.QueuedAt = nil
	effect.LeaseOwner = claim.WorkerID
	effect.LeaseToken = leaseToken
	effect.LeaseExpiresAt = &expires
	result.Effect, result.Claimed, result.SelectedResponderAgentID = effect, true, candidate
	return nil
}