package run

import (
	"context"
	"crypto/rand"
	"crypto/sha256"
	"encoding/hex"
	"fmt"
	"math/big"
	mrand "math/rand"
	"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"
)

// responderCandidatePoolKey returns the stable 64-char hex SHA-256 key of an
// ordered candidate responder pool. FIFO waiter fairness is scoped per exact
// candidate pool (oldest waiter for the same pool wakes first): two pools with
// the same ordered canonical agent IDs share one key. The fixed 64-char length
// keeps the pool_seq index within TiDB's 3072-byte key limit.
func responderCandidatePoolKey(agentIDs []string) string {
	trimmed := make([]string, 0, len(agentIDs))
	for _, id := range agentIDs {
		trimmed = append(trimmed, strings.TrimSpace(id))
	}
	sum := sha256.Sum256([]byte(strings.Join(trimmed, "\x00")))
	return hex.EncodeToString(sum[:])
}

// normalizeCandidateResponderPool trims and de-duplicates an ordered candidate
// pool in place. An empty pool is invalid for selectOne.
func normalizeCandidateResponderPool(agentIDs []string) ([]string, error) {
	seen := make(map[string]struct{}, len(agentIDs))
	ordered := make([]string, 0, len(agentIDs))
	for _, id := range agentIDs {
		trimmed := strings.TrimSpace(id)
		if trimmed == "" {
			continue
		}
		if _, exists := seen[trimmed]; exists {
			continue
		}
		seen[trimmed] = struct{}{}
		ordered = append(ordered, trimmed)
	}
	if len(ordered) == 0 {
		return nil, storeerr.BadRequest("responder selectOne candidate pool must not be empty")
	}
	return ordered, nil
}

// parseStoredSelectOneStrategy is the closed validator for a persisted waiter
// strategy read from storage. Only omitted/ordered/random are valid; a tampered
// or foreign value fails closed with a bounded error instead of silently
// becoming the legacy ordered default.
func parseStoredSelectOneStrategy(stored string) (plan.SelectOneStrategy, error) {
	if strings.TrimSpace(stored) != stored {
		return "", storeerr.BadRequest("persisted responder strategy is not canonical")
	}
	return plan.ParseSelectOneStrategy(stored)
}

// validateSelectOneStrategyParam closes the strategy boundary at the store
// entry: an unknown value fails (never normalizing to ordered) and a random pool
// needs at least two deduplicated candidates.
func validateSelectOneStrategyParam(raw plan.SelectOneStrategy, candidates []string) (plan.SelectOneStrategy, error) {
	strategy, err := plan.ParseSelectOneStrategy(string(raw))
	if err != nil {
		return "", storeerr.BadRequest(err.Error())
	}
	if strategy.Normalized() == plan.SelectOneStrategyRandom && len(candidates) < 2 {
		return "", storeerr.BadRequest("responder selectOne strategy \"random\" requires at least two deduplicated candidates")
	}
	return strategy, nil
}

// SelectCallbackResponderCapacity is the selectOne variant of
// ClaimCallbackEffectCapacity. Given an ordered candidate responder pool, it
// atomically selects the highest-priority candidate that currently has free
// callback capacity, claims its (workspace, Agent) execution lease, freezes it
// as the callback responder, and returns Claimed so the caller delivers now.
// When EVERY candidate is busy, the effect is durably parked as
// waiting_for_responder (it is NOT bound to any candidate and does NOT occupy
// any candidate's FIFO queue) with a FIFO waiter position, and the caller must
// NOT deliver. The effect is later promoted by PromoteCallbackResponderWaiters.
//
// Callers must already hold the effect's dispatch lease via ClaimEffect; this
// method revalidates it inside its transaction so a stale or reclaimed worker
// can never misapply a candidate-pool decision. Lock order matches the
// single-responder claim:
//
//	placement -> Run -> Effect -> candidate execution leases -> callback queue
type SelectCallbackResponderCapacityParams struct {
	WorkspaceID  string
	RunID        string
	ActivationID string
	EffectID     string
	WorkerID     string
	NodeID       string
	Now          time.Time
	LeaseTTL     time.Duration
	LeaseToken   string
	// CandidateResponderAgentIDs is the ordered, priority-first candidate pool.
	CandidateResponderAgentIDs []string
	// SelectOneStrategy is the additive responder.selectOne strategy. Empty or
	// "ordered" keeps the legacy strict-priority selection; "random" picks
	// uniformly at random among the currently available deduplicated candidates
	// using Randomizer (defaulting to a crypto-backed source when nil).
	SelectOneStrategy plan.SelectOneStrategy
	// Randomizer is the injectable unbiased source for the "random" strategy.
	// It is used only when SelectOneStrategy is "random"; a nil value falls back
	// to a crypto/rand-backed source so production is never predictable. An
	// injected Randomizer that returns a bounded error proves the crypto-failure
	// path fails closed (selects nobody, persists nothing).
	Randomizer Randomizer
}

// Randomizer draws an unbiased index in [0,n) for the "random" selectOne
// strategy. It returns a bounded error on failure (never a fallback index), so
// a crypto/rand failure fails closed instead of silently selecting candidate 0.
type Randomizer interface {
	// Draw returns a uniform index in [0, n), or a bounded error. n must be > 0.
	Draw(n int) (int, error)
}

// cryptoRandomizer is the production default backed by crypto/rand.
type cryptoRandomizer struct{}

func (cryptoRandomizer) Draw(n int) (int, error) {
	if n <= 0 {
		return 0, fmt.Errorf("selectOne random pool is empty")
	}
	value, err := rand.Int(rand.Reader, big.NewInt(int64(n)))
	if err != nil {
		return 0, fmt.Errorf("crypto random source failed: %w", err)
	}
	return int(value.Int64()), nil
}

// seededRandomizer adapts a *mrand.Rand to the Randomizer seam for deterministic
// tests.
type seededRandomizer struct {
	source *mrand.Rand
}

func (r seededRandomizer) Draw(n int) (int, error) {
	if n <= 0 {
		return 0, fmt.Errorf("selectOne random pool is empty")
	}
	return r.source.Intn(n), nil
}

// NewSeededRandomizer returns a deterministic, injectable Randomizer for tests.
func NewSeededRandomizer(seed int64) Randomizer {
	return seededRandomizer{source: mrand.New(mrand.NewSource(seed))}
}

// selectOneRandomizer returns the effective draw for a selectOne random
// strategy: the injected source when present, otherwise a crypto-backed one.
// The crypto path returns an error (never index 0) so a random-source failure
// fails closed instead of silently biasing to the first candidate.
func selectOneRandomizer(randomizer Randomizer) func(int) (int, error) {
	if randomizer == nil {
		randomizer = cryptoRandomizer{}
	}
	return randomizer.Draw
}

// selectOneRandomAvailableCandidate implements the additive "random" selectOne
// strategy. It gathers every currently available deduplicated candidate (free
// callback capacity and not reserved by an older waiter) and picks one
// uniformly at random with the injectable unbiased randomizer. waited is true
// when no candidate is available (the caller parks the effect waiting).
func selectOneRandomAvailableCandidate(
	ctx context.Context,
	tx coredb.Tx,
	store *Store,
	params SelectCallbackResponderCapacityParams,
	candidates []string,
) (string, bool, error) {
	available := make([]string, 0, len(candidates))
	for _, candidate := range candidates {
		lease, leaseFound, err := store.lockAgentExecutionLease(ctx, tx, params.WorkspaceID, candidate)
		if err != nil {
			return "", false, err
		}
		free, err := candidateFreeCapacityLocked(ctx, tx, params.WorkspaceID, candidate, params.EffectID, lease, leaseFound)
		if err != nil {
			return "", false, err
		}
		if !free {
			continue
		}
		reserved, err := store.candidateReservedByOlderWaiter(ctx, tx, params.WorkspaceID, candidate, params.EffectID)
		if err != nil {
			return "", false, err
		}
		if reserved {
			continue
		}
		available = append(available, candidate)
	}
	if len(available) == 0 {
		return "", true, nil
	}
	draw, err := selectOneRandomizer(params.Randomizer)(len(available))
	if err != nil {
		// A crypto random-source failure must fail closed (never index 0).
		return "", false, err
	}
	return available[draw], false, nil
}

// SelectCallbackResponderCapacityResult reports the durable candidate-pool
// decision. Claimed means the caller must deliver now to
// SelectedResponderAgentID. Waiting means the effect is durably parked
// waiting_for_responder and the caller must NOT deliver.
type SelectCallbackResponderCapacityResult struct {
	Effect                   Effect
	Claimed                  bool
	SelectedResponderAgentID string
	Waiting                  bool
}

func (store *Store) SelectCallbackResponderCapacity(
	ctx context.Context,
	params SelectCallbackResponderCapacityParams,
) (SelectCallbackResponderCapacityResult, error) {
	if err := store.Validate(); err != nil {
		return SelectCallbackResponderCapacityResult{}, err
	}
	params.WorkspaceID, params.RunID, params.EffectID = strings.TrimSpace(params.WorkspaceID), strings.TrimSpace(params.RunID), strings.TrimSpace(params.EffectID)
	params.WorkerID, params.LeaseToken, params.NodeID, params.ActivationID =
		strings.TrimSpace(params.WorkerID), strings.TrimSpace(params.LeaseToken), strings.TrimSpace(params.NodeID), strings.TrimSpace(params.ActivationID)
	if params.WorkspaceID == "" || params.RunID == "" || params.ActivationID == "" || params.EffectID == "" ||
		params.WorkerID == "" || params.LeaseToken == "" || params.LeaseTTL <= 0 {
		return SelectCallbackResponderCapacityResult{}, storeerr.BadRequest("workspaceId, runId, activationId, effectId, workerId, dispatch lease, and positive leaseTTL are required")
	}
	candidates, err := normalizeCandidateResponderPool(params.CandidateResponderAgentIDs)
	if err != nil {
		return SelectCallbackResponderCapacityResult{}, err
	}
	strategy, err := validateSelectOneStrategyParam(params.SelectOneStrategy, candidates)
	if err != nil {
		return SelectCallbackResponderCapacityResult{}, err
	}
	params.SelectOneStrategy = strategy
	for _, candidate := range candidates {
		if len(candidate) > 255 {
			return SelectCallbackResponderCapacityResult{}, storeerr.BadRequest("candidate responder agent id is too long")
		}
	}
	if params.Now.IsZero() {
		params.Now = time.Now()
	}
	params.Now = utc(params.Now)

	var result SelectCallbackResponderCapacityResult
	err = coredb.WithTx(ctx, store.db, func(tx coredb.Tx) error {
		if _, err := store.lockCandidatePlacement(ctx, tx, params.WorkspaceID); err != nil {
			return err
		}
		run, err := scanRun(tx.QueryRow(ctx, runSelectSQL()+`
WHERE r.workspace_id=$1 AND r.id=$2 FOR UPDATE`, params.WorkspaceID, params.RunID))
		switch {
		case coredb.IsNotFound(err):
			return nil
		case err != nil:
			return fmt.Errorf("lock workflow Run for callback capacity: %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`, params.WorkspaceID, params.RunID, params.EffectID))
		if err != nil {
			return mapReadError("lock workflow Run effect for callback capacity", err)
		}
		// Revalidate our dispatch lease: the effect must still be dispatching
		// under this worker's token before any capacity decision is applied.
		if effect.Status != EffectStatusDispatching || effect.LeaseToken != params.LeaseToken {
			return nil
		}
		// If the effect already froze a responder (retry/recovery of a
		// previously-selected selectOne effect), reuse it — the persisted frozen
		// responder is the sole authority and is never re-resolved.
		if frozen := strings.TrimSpace(effect.CallbackResponderAgentID); frozen != "" {
			return freezeSelectOneConfirmedDispatch(ctx, tx, &result, effect, params, frozen)
		}
		// Iterate candidates in priority order. A candidate may be claimed only
		// if it has free capacity AND no OLDER live waiting member is currently
		// eligible (FIFO arbitration, same atomic placement lock+snapshot). The
		// additive "random" strategy instead picks uniformly at random among the
		// currently available deduplicated candidates (see
		// selectOneRandomAvailableCandidate); the legacy ordered (omitted)
		// strategy remains the strict-priority selection below.
		if params.SelectOneStrategy.Normalized() == plan.SelectOneStrategyRandom {
			chosen, waited, err := selectOneRandomAvailableCandidate(ctx, tx, store, params, candidates)
			if err != nil {
				return err
			}
			if waited {
				return waitForSelectOneResponder(ctx, tx, store, &result, effect, params, candidates, run)
			}
			return selectSelectOneCandidate(ctx, tx, store, &result, effect, params, chosen, run)
		}
		for _, candidate := range candidates {
			lease, leaseFound, err := store.lockAgentExecutionLease(ctx, tx, params.WorkspaceID, candidate)
			if err != nil {
				return err
			}
			free, err := candidateFreeCapacityLocked(ctx, tx, params.WorkspaceID, candidate, params.EffectID, lease, leaseFound)
			if err != nil {
				return err
			}
			if !free {
				continue
			}
			reserved, err := store.candidateReservedByOlderWaiter(ctx, tx, params.WorkspaceID, candidate, params.EffectID)
			if err != nil {
				return err
			}
			if reserved {
				// An older waiter is now eligible for this candidate; the newcomer
				// must not steal it. Try the next (lower-priority) candidate.
				continue
			}
			return selectSelectOneCandidate(ctx, tx, store, &result, effect, params, candidate, run)
		}
		// Every candidate is busy: durably park as waiting_for_responder with a
		// FIFO waiter position for the candidate pool. The effect is NOT bound
		// to any candidate and does NOT occupy any candidate's FIFO queue.
		return waitForSelectOneResponder(ctx, tx, store, &result, effect, params, candidates, run)
	})
	if err != nil {
		return SelectCallbackResponderCapacityResult{}, err
	}
	return result, nil
}

// freezeSelectOneConfirmedDispatch re-confirms an already-frozen selectOne
// callback (retry/recovery) under the same dispatch lease, keeping the frozen
// responder and its FIFO position.
func freezeSelectOneConfirmedDispatch(
	ctx context.Context,
	tx coredb.Tx,
	result *SelectCallbackResponderCapacityResult,
	effect Effect,
	params SelectCallbackResponderCapacityParams,
	frozen string,
) error {
	claim := ClaimCallbackEffectCapacityParams{
		WorkspaceID: params.WorkspaceID, RunID: params.RunID, ActivationID: params.ActivationID,
		EffectID: params.EffectID, WorkerID: params.WorkerID, Now: params.Now,
		LeaseTTL: params.LeaseTTL, LeaseToken: params.LeaseToken, ResponderAgentID: frozen, NodeID: params.NodeID,
	}
	inner := ClaimCallbackEffectCapacityResult{}
	if err := freezeAndConfirmCallbackDispatch(ctx, tx, &inner, effect, claim, callbackQueueStatusDispatching); err != nil {
		return err
	}
	result.Effect, result.Claimed, result.SelectedResponderAgentID = inner.Effect, true, frozen
	return nil
}

// selectSelectOneCandidate acquires the chosen candidate's execution lease,
// freezes it as the callback responder, and confirms dispatch.
func selectSelectOneCandidate(
	ctx context.Context,
	tx coredb.Tx,
	store *Store,
	result *SelectCallbackResponderCapacityResult,
	effect Effect,
	params SelectCallbackResponderCapacityParams,
	candidate string,
	run Run,
) error {
	claim := ClaimCallbackEffectCapacityParams{
		WorkspaceID: params.WorkspaceID, RunID: params.RunID, ActivationID: params.ActivationID,
		EffectID: params.EffectID, WorkerID: params.WorkerID, Now: params.Now,
		LeaseTTL: params.LeaseTTL, LeaseToken: params.LeaseToken, ResponderAgentID: candidate, NodeID: params.NodeID,
	}
	inner := ClaimCallbackEffectCapacityResult{}
	if err := acquireCallbackExecutionLease(ctx, tx, &inner, effect, claim, callbackQueueStatusDispatching); err != nil {
		return err
	}
	if err := store.appendResponderCapacityEvent(ctx, tx, run, effect, params.NodeID, EventTypeEffectResponderSelected, params.Now); err != nil {
		return err
	}
	result.Effect, result.Claimed, result.SelectedResponderAgentID = inner.Effect, true, candidate
	return nil
}

// waitForSelectOneResponder transitions the dispatching effect to
// waiting_for_responder (releasing the dispatch lease) and inserts/updates its
// FIFO candidate-pool waiter row. It runs under the placement lock, so the
// waiter_seq assignment for one pool serializes across concurrent waiters.
func waitForSelectOneResponder(
	ctx context.Context,
	tx coredb.Tx,
	store *Store,
	result *SelectCallbackResponderCapacityResult,
	effect Effect,
	params SelectCallbackResponderCapacityParams,
	candidates []string,
	run Run,
) error {
	now := params.Now
	if _, err := tx.Exec(ctx, `
UPDATE workflow_run_effects
SET status='waiting_for_responder',
    lease_owner=NULL, lease_token=NULL, lease_expires_at=NULL,
    queued_at=COALESCE(queued_at,$4), updated_at=$4
WHERE workspace_id=$1 AND run_id=$2 AND id=$3 AND lease_token=$5 AND status='dispatching'`,
		params.WorkspaceID, params.RunID, params.EffectID, now, params.LeaseToken); err != nil {
		return fmt.Errorf("park workflow Run callback effect waiting for responder: %w", err)
	}
	if _, err := ensureSelectOneWaiter(ctx, tx, params, candidates, ResponderWaiterStatusWaiting, now); err != nil {
		return err
	}
	if err := store.appendResponderCapacityEvent(ctx, tx, run, effect, params.NodeID, EventTypeEffectWaitingForResponderCapacity, now); err != nil {
		return err
	}
	effect.Status = EffectStatusWaitingForResponder
	effect.LeaseOwner, effect.LeaseToken, effect.LeaseExpiresAt = "", "", nil
	effect.QueuedAt = &now
	result.Effect = effect
	result.Waiting = true
	return nil
}

// syncSelectOneMembership rewrites the per-candidate FIFO membership index rows
// for one waiter's pool. The promotion scan reads waiters through this index
// (one row per (workspace, candidate, effect)), so a freed candidate can always
// find the OLDEST live waiting waiter that includes it — no waiter cursor, no
// rotation, and no order-changing scan.
func syncSelectOneMembership(
	ctx context.Context,
	tx coredb.Tx,
	workspaceID, effectID string,
	candidates []string,
	globalSeq int64,
	now time.Time,
) error {
	if _, err := tx.Exec(ctx, `
DELETE FROM workflow_responder_candidate_members
WHERE workspace_id=$1 AND effect_id=$2`, workspaceID, effectID); err != nil {
		return fmt.Errorf("clear selectOne waiter membership: %w", err)
	}
	if len(candidates) == 0 {
		return nil
	}
	values := make([]string, 0, len(candidates))
	args := make([]any, 0, len(candidates)*6)
	for _, candidate := range candidates {
		first := len(args) + 1
		values = append(values, fmt.Sprintf("($%d,$%d,$%d,$%d,$%d,$%d)", first, first+1, first+2, first+3, first+4, first+5))
		args = append(args, workspaceID, candidate, effectID, globalSeq, now, now)
	}
	if _, err := tx.Exec(ctx, `
INSERT INTO workflow_responder_candidate_members
  (workspace_id, candidate_agent_id, effect_id, global_seq, created_at, updated_at)
VALUES `+strings.Join(values, ","), args...); err != nil {
		return fmt.Errorf("insert selectOne waiter membership: %w", err)
	}
	return nil
}

// deleteSelectOneMembership removes all membership index rows for one waiter
// when it leaves the waiting set (selected, canceled, completed). The effect is
// no longer a candidate for promotion, so it must not be returned by the
// per-candidate oldest-member scan.
func deleteSelectOneMembership(ctx context.Context, tx coredb.Tx, workspaceID, effectID string) error {
	if _, err := tx.Exec(ctx, `
DELETE FROM workflow_responder_candidate_members
WHERE workspace_id=$1 AND effect_id=$2`, workspaceID, effectID); err != nil {
		return fmt.Errorf("clear selectOne waiter membership: %w", err)
	}
	return nil
}

// ensureSelectOneWaiter inserts or updates the durable FIFO waiter row for a
// waiting_for_responder selectOne effect. The unique (workspace_id, effect_id)
// key freezes the first wait; waiter_seq is monotonic within
// (workspace_id, candidate_pool_key) so older waiters for the same pool wake
// first.
func ensureSelectOneWaiter(
	ctx context.Context,
	tx coredb.Tx,
	params SelectCallbackResponderCapacityParams,
	candidates []string,
	status string,
	now time.Time,
) (ResponderCandidateWaiter, error) {
	poolKey := responderCandidatePoolKey(candidates)
	poolJSON, err := model.MarshalJSONObject("workflow responder candidate pool", candidates)
	if err != nil {
		return ResponderCandidateWaiter{}, fmt.Errorf("encode candidate pool: %w", err)
	}
	var waiter ResponderCandidateWaiter
	var storedStrategy string
	err = tx.QueryRow(ctx, `
SELECT id, workspace_id, run_id, activation_id, effect_id, node_id,
       candidate_pool_key, candidate_pool_json, select_one_strategy, waiter_seq, global_seq, status, created_at, updated_at
FROM workflow_responder_candidate_waiters
WHERE workspace_id=$1 AND effect_id=$2`, params.WorkspaceID, params.EffectID).Scan(
		&waiter.ID, &waiter.WorkspaceID, &waiter.RunID, &waiter.ActivationID, &waiter.EffectID, &waiter.NodeID,
		&waiter.CandidatePoolKey, &waiter.CandidateAgentIDs, &storedStrategy, &waiter.WaiterSeq, &waiter.GlobalSeq, &waiter.Status, &waiter.CreatedAt, &waiter.UpdatedAt)
	if err == nil {
		parsedStrategy, parseErr := parseStoredSelectOneStrategy(storedStrategy)
		if parseErr != nil {
			return ResponderCandidateWaiter{}, parseErr
		}
		waiter.SelectOneStrategy = parsedStrategy
		if waiter.CandidatePoolKey != poolKey {
			// A re-selection changed the pool; keep the existing position but
			// refresh the pool. (Retry of a previously-selected effect never
			// reaches here because the frozen responder path reuses it.)
			if _, err := tx.Exec(ctx, `
UPDATE workflow_responder_candidate_waiters
SET candidate_pool_key=$3, candidate_pool_json=$4, updated_at=$5
WHERE workspace_id=$1 AND effect_id=$2`, params.WorkspaceID, params.EffectID, poolKey, poolJSON, now); err != nil {
				return ResponderCandidateWaiter{}, fmt.Errorf("refresh selectOne waiter pool: %w", err)
			}
		}
		if waiter.Status != status && (waiter.Status == ResponderWaiterStatusWaiting || waiter.Status == ResponderWaiterStatusSelected) {
			if _, err := tx.Exec(ctx, `
UPDATE workflow_responder_candidate_waiters
SET status=$3, updated_at=$4
WHERE workspace_id=$1 AND effect_id=$2`, params.WorkspaceID, params.EffectID, status, now); err != nil {
				return ResponderCandidateWaiter{}, fmt.Errorf("update selectOne waiter status: %w", err)
			}
		}
		waiter.Status = status
		waiter.UpdatedAt = now
		if err := syncSelectOneMembership(ctx, tx, waiter.WorkspaceID, waiter.EffectID, candidates, waiter.GlobalSeq, now); err != nil {
			return ResponderCandidateWaiter{}, err
		}
		return waiter, nil
	}
	if !coredb.IsNotFound(err) {
		return ResponderCandidateWaiter{}, fmt.Errorf("load selectOne waiter: %w", err)
	}
	var nextSeq int64
	if err := tx.QueryRow(ctx, `
SELECT COALESCE(MAX(waiter_seq), 0) + 1
FROM workflow_responder_candidate_waiters
WHERE workspace_id=$1 AND candidate_pool_key=$2`, params.WorkspaceID, poolKey).Scan(&nextSeq); err != nil {
		return ResponderCandidateWaiter{}, fmt.Errorf("assign selectOne waiter sequence: %w", err)
	}
	// global_seq is the workspace-monotonic arrival order for the promotion
	// scan. It is assigned under the placement lock (the caller holds it), so
	// concurrent waiters in the same workspace serialize and never collide.
	var nextGlobalSeq int64
	if err := tx.QueryRow(ctx, `
SELECT COALESCE(MAX(global_seq), 0) + 1
FROM workflow_responder_candidate_waiters
WHERE workspace_id=$1`, params.WorkspaceID).Scan(&nextGlobalSeq); err != nil {
		return ResponderCandidateWaiter{}, fmt.Errorf("assign selectOne global waiter sequence: %w", err)
	}
	waiterID, err := identifiers.NewEntityID("wrcw")
	if err != nil {
		return ResponderCandidateWaiter{}, fmt.Errorf("generate selectOne waiter id: %w", err)
	}
	if _, err := tx.Exec(ctx, `
INSERT INTO workflow_responder_candidate_waiters (
  id, workspace_id, run_id, activation_id, effect_id, node_id,
  candidate_pool_key, candidate_pool_json, select_one_strategy, waiter_seq, global_seq, status, created_at, updated_at
) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$13)
ON DUPLICATE KEY UPDATE
  status = IF(status IN ('waiting','selected'), VALUES(status), status),
  candidate_pool_key = VALUES(candidate_pool_key),
  candidate_pool_json = VALUES(candidate_pool_json),
  select_one_strategy = VALUES(select_one_strategy),
  updated_at = VALUES(updated_at)`,
		waiterID, params.WorkspaceID, params.RunID, params.ActivationID, params.EffectID, params.NodeID,
		poolKey, poolJSON, string(params.SelectOneStrategy.Normalized()), nextSeq, nextGlobalSeq, status, now); err != nil {
		return ResponderCandidateWaiter{}, fmt.Errorf("insert selectOne waiter: %w", err)
	}
	waiter = ResponderCandidateWaiter{
		ID: waiterID, WorkspaceID: params.WorkspaceID, RunID: params.RunID, ActivationID: params.ActivationID,
		EffectID: params.EffectID, NodeID: params.NodeID, CandidatePoolKey: poolKey,
		CandidateAgentIDs: candidates, SelectOneStrategy: params.SelectOneStrategy.Normalized(),
		WaiterSeq: nextSeq, GlobalSeq: nextGlobalSeq, Status: status, CreatedAt: now, UpdatedAt: now,
	}
	if err := syncSelectOneMembership(ctx, tx, waiter.WorkspaceID, waiter.EffectID, candidates, nextGlobalSeq, now); err != nil {
		return ResponderCandidateWaiter{}, err
	}
	return waiter, nil
}

// ListCallbackResponderWaitersParams bounds a read of candidate-pool waiters.
type ListCallbackResponderWaitersParams struct {
	WorkspaceID string
	// CandidatePoolKey optionally filters to one exact ordered pool; empty means
	// every pool in the workspace.
	CandidatePoolKey string
	AfterWaiterSeq   int64
	Limit            int
}

// ListCallbackResponderWaiters returns the durable FIFO candidate-pool waiters
// for a workspace, oldest first, bounded for read-model projection.
func (store *Store) ListCallbackResponderWaiters(ctx context.Context, params ListCallbackResponderWaitersParams) ([]ResponderCandidateWaiter, error) {
	if err := store.Validate(); err != nil {
		return nil, err
	}
	params.WorkspaceID = strings.TrimSpace(params.WorkspaceID)
	if params.WorkspaceID == "" {
		return nil, storeerr.BadRequest("workspaceId is required")
	}
	bounded, err := boundedLimit(params.Limit)
	if err != nil {
		return nil, err
	}
	poolFilter := ""
	args := []any{params.WorkspaceID, params.AfterWaiterSeq, bounded}
	if strings.TrimSpace(params.CandidatePoolKey) != "" {
		poolFilter = " AND candidate_pool_key=$4"
		args = append(args, params.CandidatePoolKey)
	}
	rows, err := store.db.Query(ctx, `
SELECT id, workspace_id, run_id, activation_id, effect_id, node_id,
       candidate_pool_key, candidate_pool_json, select_one_strategy, waiter_seq, global_seq, status, created_at, updated_at
FROM workflow_responder_candidate_waiters
WHERE workspace_id=$1 AND waiter_seq>$2 AND status IN ('waiting','selected')`+poolFilter+`
ORDER BY candidate_pool_key ASC, waiter_seq ASC
LIMIT $3`, args...)
	if err != nil {
		return nil, fmt.Errorf("list workflow candidate-pool waiters: %w", err)
	}
	defer rows.Close()
	waiters := make([]ResponderCandidateWaiter, 0, bounded)
	for rows.Next() {
		var waiter ResponderCandidateWaiter
		var poolJSON []byte
		var storedStrategy string
		if err := rows.Scan(&waiter.ID, &waiter.WorkspaceID, &waiter.RunID, &waiter.ActivationID, &waiter.EffectID, &waiter.NodeID,
			&waiter.CandidatePoolKey, &poolJSON, &storedStrategy, &waiter.WaiterSeq, &waiter.GlobalSeq, &waiter.Status, &waiter.CreatedAt, &waiter.UpdatedAt); err != nil {
			return nil, fmt.Errorf("scan workflow candidate-pool waiter: %w", err)
		}
		if err := model.UnmarshalJSONObject("workflow responder candidate pool", poolJSON, &waiter.CandidateAgentIDs); err != nil {
			return nil, fmt.Errorf("decode workflow candidate-pool waiter: %w", err)
		}
		parsedStrategy, parseErr := parseStoredSelectOneStrategy(storedStrategy)
		if parseErr != nil {
			return nil, parseErr
		}
		waiter.SelectOneStrategy = parsedStrategy
		waiters = append(waiters, waiter)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("iterate workflow candidate-pool waiters: %w", err)
	}
	return waiters, nil
}

// LoadCallbackResponderWaiter returns the durable candidate-pool waiter for one
// waiting_for_responder effect, or ok=false when the effect never waited.
func (store *Store) LoadCallbackResponderWaiter(ctx context.Context, workspaceID, effectID string) (ResponderCandidateWaiter, bool, error) {
	if err := store.Validate(); err != nil {
		return ResponderCandidateWaiter{}, false, err
	}
	workspaceID, effectID = strings.TrimSpace(workspaceID), strings.TrimSpace(effectID)
	if workspaceID == "" || effectID == "" {
		return ResponderCandidateWaiter{}, false, storeerr.BadRequest("workspaceId and effectId are required")
	}
	var waiter ResponderCandidateWaiter
	var poolJSON []byte
	var storedStrategy string
	err := store.db.QueryRow(ctx, `
SELECT id, workspace_id, run_id, activation_id, effect_id, node_id,
       candidate_pool_key, candidate_pool_json, select_one_strategy, waiter_seq, global_seq, status, created_at, updated_at
FROM workflow_responder_candidate_waiters
WHERE workspace_id=$1 AND effect_id=$2`, workspaceID, effectID).Scan(
		&waiter.ID, &waiter.WorkspaceID, &waiter.RunID, &waiter.ActivationID, &waiter.EffectID, &waiter.NodeID,
		&waiter.CandidatePoolKey, &poolJSON, &storedStrategy, &waiter.WaiterSeq, &waiter.GlobalSeq, &waiter.Status, &waiter.CreatedAt, &waiter.UpdatedAt)
	if coredb.IsNotFound(err) {
		return ResponderCandidateWaiter{}, false, nil
	}
	if err != nil {
		return ResponderCandidateWaiter{}, false, fmt.Errorf("load workflow candidate-pool waiter: %w", err)
	}
	if err := model.UnmarshalJSONObject("workflow responder candidate pool", poolJSON, &waiter.CandidateAgentIDs); err != nil {
		return ResponderCandidateWaiter{}, false, fmt.Errorf("decode workflow candidate-pool waiter: %w", err)
	}
	parsedStrategy, parseErr := parseStoredSelectOneStrategy(storedStrategy)
	if parseErr != nil {
		return ResponderCandidateWaiter{}, false, parseErr
	}
	waiter.SelectOneStrategy = parsedStrategy
	return waiter, true, nil
}

// candidateFreeCapacityLocked reports whether a candidate currently has free
// callback capacity under the caller's transaction (its lease row already
// locked FOR UPDATE by the caller). It mirrors the single-responder busy check:
// a held lease by another effect, a legacy awaiting barrier, or queued items
// ahead all make the candidate busy.
func candidateFreeCapacityLocked(
	ctx context.Context,
	tx coredb.Tx,
	workspaceID, candidate, effectID string,
	lease AgentExecutionLease,
	leaseFound bool,
) (bool, error) {
	if leaseFound && lease.Status == agentExecutionLeaseStatusHeld && lease.EffectID != effectID {
		return false, nil
	}
	if leaseFound && lease.Status == agentExecutionLeaseStatusHeld && lease.EffectID == effectID {
		// Already the owner (retry); treat as free so promotion can re-confirm.
		return true, nil
	}
	barrier, err := legacyAwaitingBarrierForAgent(ctx, tx, workspaceID, candidate, effectID)
	if err != nil {
		return false, err
	}
	if barrier {
		return false, nil
	}
	ahead, err := callbackQueueAheadCount(ctx, tx, workspaceID, candidate, effectID)
	if err != nil {
		return false, err
	}
	return ahead == 0, nil
}