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/storeerr"
)

// AgentExecutionLease is the durable capacity=1 execution lease for one
// (workspace, canonical Agent). A held lease proves that exactly one Agent
// callback is executing or awaiting its reply for that workspace/Agent pair;
// every other callback for the pair is queued behind it. The lease expresses
// Workflow capacity/ordering only — it is NEVER read as Agent runtime liveness
// by the automatic-recovery working signal.
type AgentExecutionLease struct {
	WorkspaceID      string
	ResponderAgentID string
	RunID            string
	ActivationID     string
	EffectID         string
	NodeID           string
	LeaseToken       string
	Status           string
	CreatedAt        time.Time
	UpdatedAt        time.Time
}

// AgentCallbackQueueItem is the stable FIFO position of one callback effect for
// one (workspace, Agent). enqueue_seq is a persistent monotonic order assigned
// under the execution-lease lock, so the first enqueue is frozen and retries,
// restarts, renames, or membership changes never move an item.
type AgentCallbackQueueItem struct {
	ID               string
	WorkspaceID      string
	RunID            string
	ActivationID     string
	EffectID         string
	ResponderAgentID string
	NodeID           string
	EnqueueSeq       int64
	Status           string
	CreatedAt        time.Time
	UpdatedAt        time.Time
}

const (
	agentExecutionLeaseStatusHeld     = "held"
	agentExecutionLeaseStatusReleased = "released"

	callbackQueueStatusQueued      = "queued"
	callbackQueueStatusDispatching = "dispatching"
	callbackQueueStatusCompleted   = "completed"
	callbackQueueStatusCanceled    = "canceled"
)

// ClaimCallbackEffectCapacityParams freezes the canonical callback responder
// and atomically acquires the (workspace, Agent) execution lease or moves the
// effect into the stable FIFO queue. Callers must have already claimed the
// effect's dispatch lease via ClaimEffect; this method revalidates that lease
// inside its transaction so a stale or reclaimed worker can never misapply a
// capacity decision.
type ClaimCallbackEffectCapacityParams struct {
	WorkspaceID      string
	RunID            string
	ActivationID     string
	EffectID         string
	WorkerID         string
	Now              time.Time
	LeaseTTL         time.Duration
	LeaseToken       string
	ResponderAgentID string
	NodeID           string
}

// ClaimCallbackEffectCapacityResult reports the durable capacity decision.
// Claimed means the effect keeps its dispatch lease and the caller must
// deliver the callback message now. Queued means the effect is durably
// queued_for_agent and the caller must not deliver.
type ClaimCallbackEffectCapacityResult struct {
	Effect  Effect
	Claimed bool
	Queued  bool
}

// ClaimCallbackEffectCapacity freezes the canonical callback Agent
// and atomically acquires the (workspace, Agent) execution lease or enqueues the
// effect in FIFO order. The lease row is locked FOR UPDATE so concurrent
// claims for the same Agent serialize; there is never a check-then-act across
// transactions. Lock order matches claims and cancellation:
//
//	placement -> Run -> Effect -> execution lease -> callback queue
func (store *Store) ClaimCallbackEffectCapacity(ctx context.Context, params ClaimCallbackEffectCapacityParams) (ClaimCallbackEffectCapacityResult, error) {
	if err := store.Validate(); err != nil {
		return ClaimCallbackEffectCapacityResult{}, err
	}
	params.WorkspaceID, params.RunID, params.EffectID = strings.TrimSpace(params.WorkspaceID), strings.TrimSpace(params.RunID), strings.TrimSpace(params.EffectID)
	params.WorkerID, params.LeaseToken, params.ResponderAgentID, params.NodeID, params.ActivationID =
		strings.TrimSpace(params.WorkerID), strings.TrimSpace(params.LeaseToken), strings.TrimSpace(params.ResponderAgentID), strings.TrimSpace(params.NodeID), strings.TrimSpace(params.ActivationID)
	if params.WorkspaceID == "" || params.RunID == "" || params.ActivationID == "" || params.EffectID == "" ||
		params.WorkerID == "" || params.LeaseToken == "" || params.ResponderAgentID == "" || params.LeaseTTL <= 0 {
		return ClaimCallbackEffectCapacityResult{}, storeerr.BadRequest("workspaceId, runId, activationId, effectId, workerId, dispatch lease, canonical responder, and positive leaseTTL are required")
	}
	if len(params.ResponderAgentID) > 255 {
		return ClaimCallbackEffectCapacityResult{}, storeerr.BadRequest("callback responder agent id is too long")
	}
	if params.Now.IsZero() {
		params.Now = time.Now()
	}
	params.Now = utc(params.Now)
	var result ClaimCallbackEffectCapacityResult
	err := coredb.WithTx(ctx, store.db, func(tx coredb.Tx) error {
		if _, err := store.lockCandidatePlacement(ctx, tx, params.WorkspaceID); err != nil {
			return err
		}
		var runStatus string
		err := tx.QueryRow(ctx, `
SELECT status
FROM workflow_run_instances
WHERE workspace_id=$1 AND id=$2
FOR UPDATE`, params.WorkspaceID, params.RunID).Scan(&runStatus)
		switch {
		case coredb.IsNotFound(err):
			return nil
		case err != nil:
			return fmt.Errorf("lock workflow Run for callback capacity: %w", err)
		case runStatus != RunStatusRunning && runStatus != 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
		}
		// The persisted effect's frozen responder is the sole authority. A claim
		// that names a different Agent would split the effect/queue/lease
		// identity, so it fails closed instead of overwriting the grant.
		if frozen := strings.TrimSpace(effect.CallbackResponderAgentID); frozen != "" && frozen != params.ResponderAgentID {
			return storeerr.Conflict("workflow Run callback responder is already frozen to a different Agent")
		}
		lease, leaseFound, err := store.lockAgentExecutionLease(ctx, tx, params.WorkspaceID, params.ResponderAgentID)
		if err != nil {
			return err
		}
		busy := leaseFound && lease.Status == agentExecutionLeaseStatusHeld && lease.EffectID != params.EffectID
		if !busy && leaseFound && lease.Status == agentExecutionLeaseStatusHeld && lease.EffectID == params.EffectID {
			// We are already the execution owner (retry/recovery of the same
			// callback). Keep the dispatch lease and re-freeze the responder.
			return freezeAndConfirmCallbackDispatch(ctx, tx, &result, effect, params, callbackQueueStatusDispatching)
		}
		if !busy {
			// The lease is free (missing or released). A legacy awaiting receipt
			// for this Agent is still an active barrier: capacity is not released
			// until every such receipt is terminal.
			barrier, err := legacyAwaitingBarrierForAgent(ctx, tx, params.WorkspaceID, params.ResponderAgentID, params.EffectID)
			if err != nil {
				return err
			}
			if barrier {
				busy = true
			}
		}
		if busy {
			// Capacity is held by another callback or by a legacy barrier. Enqueue
			// this effect in FIFO order and release the dispatch lease.
			return queueCallbackEffect(ctx, tx, &result, effect, params, callbackQueueStatusQueued)
		}
		ahead, err := callbackQueueAheadCount(ctx, tx, params.WorkspaceID, params.ResponderAgentID, params.EffectID)
		if err != nil {
			return err
		}
		if ahead > 0 {
			// Earlier callbacks are already queued; never jump them.
			return queueCallbackEffect(ctx, tx, &result, effect, params, callbackQueueStatusQueued)
		}
		// Free capacity, no earlier queue item: acquire the execution lease and
		// dispatch now.
		return acquireCallbackExecutionLease(ctx, tx, &result, effect, params, callbackQueueStatusDispatching)
	})
	if err != nil {
		return ClaimCallbackEffectCapacityResult{}, err
	}
	return result, nil
}

func freezeAndConfirmCallbackDispatch(
	ctx context.Context,
	tx coredb.Tx,
	result *ClaimCallbackEffectCapacityResult,
	effect Effect,
	params ClaimCallbackEffectCapacityParams,
	queueStatus string,
) error {
	if _, err := tx.Exec(ctx, `
UPDATE workflow_run_effects
SET callback_responder_agent_id=$4, queued_at=NULL, updated_at=$5
WHERE workspace_id=$1 AND run_id=$2 AND id=$3 AND lease_token=$6 AND status='dispatching'`,
		params.WorkspaceID, params.RunID, params.EffectID, params.ResponderAgentID, params.Now, params.LeaseToken); err != nil {
		return fmt.Errorf("freeze callback responder: %w", err)
	}
	if _, err := ensureCallbackQueueItem(ctx, tx, params, callbackQueueStatusDispatching, params.Now); err != nil {
		return err
	}
	effect.CallbackResponderAgentID = params.ResponderAgentID
	effect.QueuedAt = nil
	result.Effect, result.Claimed = effect, true
	return nil
}

func queueCallbackEffect(
	ctx context.Context,
	tx coredb.Tx,
	result *ClaimCallbackEffectCapacityResult,
	effect Effect,
	params ClaimCallbackEffectCapacityParams,
	queueStatus string,
) error {
	now := params.Now
	if _, err := tx.Exec(ctx, `
UPDATE workflow_run_effects
SET status='queued_for_agent', callback_responder_agent_id=$4,
    queued_at=COALESCE(queued_at,$5),
    lease_owner=NULL, lease_token=NULL, lease_expires_at=NULL,
    updated_at=$5
WHERE workspace_id=$1 AND run_id=$2 AND id=$3 AND lease_token=$6 AND status='dispatching'`,
		params.WorkspaceID, params.RunID, params.EffectID, params.ResponderAgentID, now, params.LeaseToken); err != nil {
		return fmt.Errorf("queue workflow Run callback effect: %w", err)
	}
	if _, err := ensureCallbackQueueItem(ctx, tx, params, callbackQueueStatusQueued, now); err != nil {
		return err
	}
	effect.Status = EffectStatusQueuedForAgent
	effect.CallbackResponderAgentID = params.ResponderAgentID
	effect.QueuedAt = &now
	effect.LeaseOwner, effect.LeaseToken, effect.LeaseExpiresAt = "", "", nil
	result.Effect, result.Queued = effect, true
	return nil
}

func acquireCallbackExecutionLease(
	ctx context.Context,
	tx coredb.Tx,
	result *ClaimCallbackEffectCapacityResult,
	effect Effect,
	params ClaimCallbackEffectCapacityParams,
	queueStatus string,
) error {
	now := params.Now
	if err := upsertHeldCallbackLease(ctx, tx, params, now); err != nil {
		return err
	}
	if _, err := tx.Exec(ctx, `
UPDATE workflow_run_effects
SET callback_responder_agent_id=$4, queued_at=NULL, updated_at=$5
WHERE workspace_id=$1 AND run_id=$2 AND id=$3 AND lease_token=$6 AND status='dispatching'`,
		params.WorkspaceID, params.RunID, params.EffectID, params.ResponderAgentID, now, params.LeaseToken); err != nil {
		return fmt.Errorf("freeze callback responder: %w", err)
	}
	// The dispatched owner also occupies a FIFO position so later queued items
	// can count it as ahead and never jump it.
	if _, err := ensureCallbackQueueItem(ctx, tx, params, callbackQueueStatusDispatching, now); err != nil {
		return err
	}
	effect.CallbackResponderAgentID = params.ResponderAgentID
	effect.QueuedAt = nil
	result.Effect, result.Claimed = effect, true
	return nil
}

// ensureCallbackQueueItem inserts the stable FIFO position for a callback
// effect, or updates an existing one. The unique (workspace_id, effect_id) key
// freezes the first enqueue: retries and restarts never reorder or duplicate
// an item. status transitions to dispatching when a queued item is promoted.
func ensureCallbackQueueItem(ctx context.Context, tx coredb.Tx, params ClaimCallbackEffectCapacityParams, status string, now time.Time) (AgentCallbackQueueItem, error) {
	var item AgentCallbackQueueItem
	err := tx.QueryRow(ctx, `
SELECT id, workspace_id, run_id, activation_id, effect_id, responder_agent_id, node_id,
       enqueue_seq, status, created_at, updated_at
FROM workflow_agent_callback_queue
WHERE workspace_id=$1 AND effect_id=$2`, params.WorkspaceID, params.EffectID).Scan(
		&item.ID, &item.WorkspaceID, &item.RunID, &item.ActivationID, &item.EffectID, &item.ResponderAgentID,
		&item.NodeID, &item.EnqueueSeq, &item.Status, &item.CreatedAt, &item.UpdatedAt)
	if err == nil {
		if item.Status != status && (item.Status == callbackQueueStatusQueued || item.Status == callbackQueueStatusDispatching) {
			if _, err := tx.Exec(ctx, `
UPDATE workflow_agent_callback_queue
SET status=$3, updated_at=$4
WHERE workspace_id=$1 AND effect_id=$2`, params.WorkspaceID, params.EffectID, status, now); err != nil {
				return AgentCallbackQueueItem{}, fmt.Errorf("update workflow callback queue status: %w", err)
			}
		}
		return item, nil
	}
	if !coredb.IsNotFound(err) {
		return AgentCallbackQueueItem{}, fmt.Errorf("load workflow callback queue item: %w", err)
	}
	var nextSeq int64
	if err := tx.QueryRow(ctx, `
SELECT COALESCE(MAX(enqueue_seq), 0) + 1
FROM workflow_agent_callback_queue
WHERE workspace_id=$1 AND responder_agent_id=$2`, params.WorkspaceID, params.ResponderAgentID).Scan(&nextSeq); err != nil {
		return AgentCallbackQueueItem{}, fmt.Errorf("assign workflow callback queue sequence: %w", err)
	}
	itemID, err := identifiers.NewEntityID("wacq")
	if err != nil {
		return AgentCallbackQueueItem{}, fmt.Errorf("generate workflow callback queue id: %w", err)
	}
	if _, err := tx.Exec(ctx, `
INSERT INTO workflow_agent_callback_queue (
  id, workspace_id, run_id, activation_id, effect_id, responder_agent_id, node_id,
  enqueue_seq, status, created_at, updated_at
) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$10)
ON DUPLICATE KEY UPDATE
  status = IF(status IN ('queued','dispatching'), VALUES(status), status),
  updated_at = VALUES(updated_at)`,
		itemID, params.WorkspaceID, params.RunID, params.ActivationID, params.EffectID, params.ResponderAgentID, params.NodeID,
		nextSeq, status, now); err != nil {
		return AgentCallbackQueueItem{}, fmt.Errorf("insert workflow callback queue item: %w", err)
	}
	item = AgentCallbackQueueItem{
		ID: itemID, WorkspaceID: params.WorkspaceID, RunID: params.RunID, ActivationID: params.ActivationID,
		EffectID: params.EffectID, ResponderAgentID: params.ResponderAgentID, NodeID: params.NodeID,
		EnqueueSeq: nextSeq, Status: status, CreatedAt: now, UpdatedAt: now,
	}
	return item, nil
}

func updateCallbackQueueStatus(ctx context.Context, tx coredb.Tx, workspaceID, effectID, status string, now time.Time) error {
	if _, err := tx.Exec(ctx, `
UPDATE workflow_agent_callback_queue
SET status=$3, updated_at=$4
WHERE workspace_id=$1 AND effect_id=$2 AND status IN ('queued','dispatching')`, workspaceID, effectID, status, now); err != nil {
		return fmt.Errorf("update workflow callback queue status: %w", err)
	}
	return nil
}

func upsertHeldCallbackLease(ctx context.Context, tx coredb.Tx, params ClaimCallbackEffectCapacityParams, now time.Time) error {
	leaseToken, err := identifiers.NewEntityID("wael")
	if err != nil {
		return fmt.Errorf("generate workflow Agent execution lease: %w", err)
	}
	if _, err := tx.Exec(ctx, `
INSERT INTO workflow_agent_execution_leases (
  workspace_id, responder_agent_id, run_id, activation_id, effect_id, node_id,
  lease_token, status, created_at, updated_at
) VALUES ($1,$2,$3,$4,$5,$6,$7,'held',$8,$8)
ON DUPLICATE KEY UPDATE
  run_id=VALUES(run_id), activation_id=VALUES(activation_id), effect_id=VALUES(effect_id),
  node_id=VALUES(node_id), lease_token=VALUES(lease_token), status=VALUES(status),
  updated_at=VALUES(updated_at)`,
		params.WorkspaceID, params.ResponderAgentID, params.RunID, params.ActivationID, params.EffectID, params.NodeID,
		leaseToken, now); err != nil {
		return fmt.Errorf("acquire workflow Agent execution lease: %w", err)
	}
	return nil
}

// lockAgentExecutionLease locks the capacity=1 lease row for one
// (workspace, Agent). The row is created lazily; a missing row is equivalent
// to released capacity.
func (store *Store) lockAgentExecutionLease(ctx context.Context, tx coredb.Tx, workspaceID, responderAgentID string) (AgentExecutionLease, bool, error) {
	var lease AgentExecutionLease
	err := tx.QueryRow(ctx, `
SELECT workspace_id, responder_agent_id, run_id, activation_id, effect_id, node_id,
       lease_token, status, created_at, updated_at
FROM workflow_agent_execution_leases
WHERE workspace_id=$1 AND responder_agent_id=$2
FOR UPDATE`, workspaceID, responderAgentID).Scan(
		&lease.WorkspaceID, &lease.ResponderAgentID, &lease.RunID, &lease.ActivationID, &lease.EffectID,
		&lease.NodeID, &lease.LeaseToken, &lease.Status, &lease.CreatedAt, &lease.UpdatedAt)
	if coredb.IsNotFound(err) {
		return AgentExecutionLease{}, false, nil
	}
	if err != nil {
		return AgentExecutionLease{}, false, fmt.Errorf("lock workflow Agent execution lease: %w", err)
	}
	return lease, true, nil
}

// legacyAwaitingBarrierForAgent reports whether any non-terminal awaiting
// callback receipt already names this (workspace, Agent). Pre-deployment
// awaiting receipts are active barriers; a new callback must not be delivered
// until every one of them is terminal.
func legacyAwaitingBarrierForAgent(ctx context.Context, tx coredb.Tx, workspaceID, responderAgentID, exceptEffectID string) (bool, error) {
	var marker int
	err := tx.QueryRow(ctx, `
SELECT 1
FROM workflow_run_effects
WHERE workspace_id=$1
  AND callback_responder_agent_id=$2
  AND callback_responder_agent_id <> ''
  AND status='awaiting'
  AND id<>$3
LIMIT 1`, workspaceID, responderAgentID, exceptEffectID).Scan(&marker)
	if coredb.IsNotFound(err) {
		return false, nil
	}
	if err != nil {
		return false, fmt.Errorf("load workflow Agent legacy awaiting barrier: %w", err)
	}
	return true, nil
}

// callbackQueueAheadCount returns the number of earlier, still-active queue
// items ahead of effectID for one (workspace, Agent). An effect with no queue
// item yet counts every active item as ahead of it, so a fresh callback can
// never jump callbacks already waiting.
func callbackQueueAheadCount(ctx context.Context, tx coredb.Tx, workspaceID, responderAgentID, effectID string) (int64, error) {
	var ahead int64
	var existingSeq int64
	var hasItem bool
	err := tx.QueryRow(ctx, `
SELECT enqueue_seq
FROM workflow_agent_callback_queue
WHERE workspace_id=$1 AND effect_id=$2`, workspaceID, effectID).Scan(&existingSeq)
	if err == nil {
		hasItem = true
	} else if !coredb.IsNotFound(err) {
		return 0, fmt.Errorf("load workflow callback queue item position: %w", err)
	}
	if hasItem {
		err = tx.QueryRow(ctx, `
SELECT COUNT(*)
FROM workflow_agent_callback_queue
WHERE workspace_id=$1 AND responder_agent_id=$2
  AND status IN ('queued','dispatching')
  AND enqueue_seq < $3`, workspaceID, responderAgentID, existingSeq).Scan(&ahead)
	} else {
		err = tx.QueryRow(ctx, `
SELECT COUNT(*)
FROM workflow_agent_callback_queue
WHERE workspace_id=$1 AND responder_agent_id=$2
  AND status IN ('queued','dispatching')`, workspaceID, responderAgentID).Scan(&ahead)
	}
	if err != nil {
		return 0, fmt.Errorf("count workflow callback queue ahead: %w", err)
	}
	return ahead, nil
}

// cancelCallbackExecutionLeases terminals every cancelable callback effect of
// one Run and releases its execution lease and FIFO queue item in the caller's
// transaction, so Run cancellation and execution-capacity release stay atomic.
// queued_for_agent callbacks are not crossed-dispatch and are cancelable;
// awaiting callbacks already hold their execution lease. The effect rows are
// locked FOR UPDATE first.
func cancelCallbackExecutionLeases(ctx context.Context, tx coredb.Tx, workspaceID, runID, errorCode string, errorJSON []byte, now time.Time) error {
	rows, err := tx.Query(ctx, `
SELECT id, COALESCE(callback_responder_agent_id,'')
FROM workflow_run_effects
WHERE workspace_id=$1 AND run_id=$2
  AND ((status='pending' AND attempt_count=0) OR status IN ('delivered','awaiting','queued_for_agent','waiting_for_responder'))
FOR UPDATE`, workspaceID, runID)
	if err != nil {
		return fmt.Errorf("lock workflow Run cancelable effects: %w", err)
	}
	type canceledCallback struct {
		effectID  string
		responder string
	}
	var callbacks []canceledCallback
	var waiting []string
	for rows.Next() {
		var callback canceledCallback
		if err := rows.Scan(&callback.effectID, &callback.responder); err != nil {
			rows.Close()
			return fmt.Errorf("scan workflow Run cancelable effect: %w", err)
		}
		callbacks = append(callbacks, callback)
		if callback.responder == "" {
			// A waiting_for_responder selectOne effect has no frozen responder
			// (no execution lease, no FIFO position); its waiter row must be
			// terminalized below.
			waiting = append(waiting, callback.effectID)
		}
	}
	rows.Close()
	if err := rows.Err(); err != nil {
		return fmt.Errorf("iterate workflow Run cancelable effects: %w", err)
	}
	if _, err := tx.Exec(ctx, `
UPDATE workflow_run_effects
SET status='canceled',retry_at=NULL,lease_owner=NULL,lease_token=NULL,lease_expires_at=NULL,
    error_code=$4,error_json=$5,completed_at=$3,updated_at=$3,
    park_state='',park_armed_at=NULL,park_parked_at=NULL,park_cleared_at=$3,
    recovery_paused_at=NULL,next_recovery_at=NULL,
    recovery_cycle_next_recovery_at=NULL
WHERE workspace_id=$1 AND run_id=$2
  AND ((status='pending' AND attempt_count=0) OR status IN ('delivered','awaiting','queued_for_agent','waiting_for_responder'))`,
		workspaceID, runID, now, errorCode, errorJSON); err != nil {
		return fmt.Errorf("cancel workflow Run callback effects: %w", err)
	}
	for _, callback := range callbacks {
		if callback.responder == "" {
			continue
		}
		if err := releaseCallbackAgentExecutionLease(ctx, tx, workspaceID, callback.responder, callback.effectID, callbackQueueStatusCanceled, now); err != nil {
			return err
		}
	}
	for _, effectID := range waiting {
		if err := terminalizeSelectOneWaiter(ctx, tx, workspaceID, effectID, ResponderWaiterStatusCanceled, now); err != nil {
			return err
		}
	}
	return nil
}

// terminalizeSelectOneWaiter marks a waiting_for_responder selectOne waiter row
// terminal (canceled/completed) when its effect is canceled or completed. It is
// called inside the caller's transaction after the effect row is already locked,
// so no cross-Run poke or extra lock ordering is introduced.
func terminalizeSelectOneWaiter(ctx context.Context, tx coredb.Tx, workspaceID, effectID, status string, now time.Time) error {
	if _, err := tx.Exec(ctx, `
UPDATE workflow_responder_candidate_waiters
SET status=$3, updated_at=$4
WHERE workspace_id=$1 AND effect_id=$2 AND status IN ('waiting','selected')`,
		workspaceID, effectID, status, now); err != nil {
		return fmt.Errorf("terminalize workflow candidate-pool waiter: %w", err)
	}
	if err := deleteSelectOneMembership(ctx, tx, workspaceID, effectID); err != nil {
		return err
	}
	return nil
}

// releaseCallbackAgentExecutionLease releases the execution lease owned by
// effectID and marks its queue item terminal. It runs inside the caller's
// transaction, which must already hold the effect's row lock, so lease release
// and the terminal effect write commit atomically. The next FIFO item is NOT
// woken here: a cross-Run UPDATE from inside this transaction would form a
// lock-order cycle with a concurrent claim on that next Run (owner completion
// waits for the next Run while the next claim waits for the Agent lease). The
// effect dispatcher polls the durable queue head on every DispatchOnce scan, so
// promotion is prompt without any cross-Run poke.
func releaseCallbackAgentExecutionLease(
	ctx context.Context,
	tx coredb.Tx,
	workspaceID, responderAgentID, effectID, queueStatus string,
	now time.Time,
) error {
	responderAgentID = strings.TrimSpace(responderAgentID)
	if responderAgentID == "" || effectID == "" {
		return nil
	}
	if _, err := tx.Exec(ctx, `
UPDATE workflow_agent_execution_leases
SET status='released', updated_at=$4
WHERE workspace_id=$1 AND responder_agent_id=$2 AND effect_id=$3 AND status='held'`,
		workspaceID, responderAgentID, effectID, now); err != nil {
		return fmt.Errorf("release workflow Agent execution lease: %w", err)
	}
	if _, err := tx.Exec(ctx, `
UPDATE workflow_agent_callback_queue
SET status=$3, updated_at=$4
WHERE workspace_id=$1 AND effect_id=$2 AND status IN ('queued','dispatching')`,
		workspaceID, effectID, queueStatus, now); err != nil {
		return fmt.Errorf("terminalize workflow callback queue item: %w", err)
	}
	return nil
}

// LoadAgentExecutionLease returns the current capacity=1 owner for one
// (workspace, Agent), or ok=false when the Agent has free capacity.
func (store *Store) LoadAgentExecutionLease(ctx context.Context, workspaceID, responderAgentID string) (AgentExecutionLease, bool, error) {
	if err := store.Validate(); err != nil {
		return AgentExecutionLease{}, false, err
	}
	workspaceID, responderAgentID = strings.TrimSpace(workspaceID), strings.TrimSpace(responderAgentID)
	if workspaceID == "" || responderAgentID == "" {
		return AgentExecutionLease{}, false, storeerr.BadRequest("workspaceId and responderAgentId are required")
	}
	var lease AgentExecutionLease
	err := store.db.QueryRow(ctx, `
SELECT workspace_id, responder_agent_id, run_id, activation_id, effect_id, node_id,
       lease_token, status, created_at, updated_at
FROM workflow_agent_execution_leases
WHERE workspace_id=$1 AND responder_agent_id=$2 AND status='held'`,
		workspaceID, responderAgentID).Scan(
		&lease.WorkspaceID, &lease.ResponderAgentID, &lease.RunID, &lease.ActivationID, &lease.EffectID,
		&lease.NodeID, &lease.LeaseToken, &lease.Status, &lease.CreatedAt, &lease.UpdatedAt)
	if coredb.IsNotFound(err) {
		return AgentExecutionLease{}, false, nil
	}
	if err != nil {
		return AgentExecutionLease{}, false, fmt.Errorf("load workflow Agent execution lease: %w", err)
	}
	return lease, true, nil
}

// ListCallbackQueueForAgent returns the FIFO queue for one (workspace, Agent),
// oldest first, bounded for read-model projection.
func (store *Store) ListCallbackQueueForAgent(ctx context.Context, workspaceID, responderAgentID string, limit int) ([]AgentCallbackQueueItem, error) {
	if err := store.Validate(); err != nil {
		return nil, err
	}
	workspaceID, responderAgentID = strings.TrimSpace(workspaceID), strings.TrimSpace(responderAgentID)
	if workspaceID == "" || responderAgentID == "" {
		return nil, storeerr.BadRequest("workspaceId and responderAgentId are required")
	}
	bounded, err := boundedLimit(limit)
	if err != nil {
		return nil, err
	}
	rows, err := store.db.Query(ctx, `
SELECT id, workspace_id, run_id, activation_id, effect_id, responder_agent_id, node_id,
       enqueue_seq, status, created_at, updated_at
FROM workflow_agent_callback_queue
WHERE workspace_id=$1 AND responder_agent_id=$2
ORDER BY enqueue_seq ASC, id ASC
LIMIT $3`, workspaceID, responderAgentID, bounded)
	if err != nil {
		return nil, fmt.Errorf("list workflow callback queue: %w", err)
	}
	defer rows.Close()
	items := make([]AgentCallbackQueueItem, 0, bounded)
	for rows.Next() {
		var item AgentCallbackQueueItem
		if err := rows.Scan(&item.ID, &item.WorkspaceID, &item.RunID, &item.ActivationID, &item.EffectID,
			&item.ResponderAgentID, &item.NodeID, &item.EnqueueSeq, &item.Status, &item.CreatedAt, &item.UpdatedAt); err != nil {
			return nil, fmt.Errorf("scan workflow callback queue item: %w", err)
		}
		items = append(items, item)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("iterate workflow callback queue: %w", err)
	}
	return items, nil
}

// LoadCallbackQueueItemForEffect returns the frozen FIFO position of one
// callback effect, or ok=false when the effect has never been queued.
func (store *Store) LoadCallbackQueueItemForEffect(ctx context.Context, workspaceID, effectID string) (AgentCallbackQueueItem, bool, error) {
	if err := store.Validate(); err != nil {
		return AgentCallbackQueueItem{}, false, err
	}
	workspaceID, effectID = strings.TrimSpace(workspaceID), strings.TrimSpace(effectID)
	if workspaceID == "" || effectID == "" {
		return AgentCallbackQueueItem{}, false, storeerr.BadRequest("workspaceId and effectId are required")
	}
	var item AgentCallbackQueueItem
	err := store.db.QueryRow(ctx, `
SELECT id, workspace_id, run_id, activation_id, effect_id, responder_agent_id, node_id,
       enqueue_seq, status, created_at, updated_at
FROM workflow_agent_callback_queue
WHERE workspace_id=$1 AND effect_id=$2`, workspaceID, effectID).Scan(
		&item.ID, &item.WorkspaceID, &item.RunID, &item.ActivationID, &item.EffectID, &item.ResponderAgentID,
		&item.NodeID, &item.EnqueueSeq, &item.Status, &item.CreatedAt, &item.UpdatedAt)
	if coredb.IsNotFound(err) {
		return AgentCallbackQueueItem{}, false, nil
	}
	if err != nil {
		return AgentCallbackQueueItem{}, false, fmt.Errorf("load workflow callback queue item: %w", err)
	}
	return item, true, nil
}

// AgentExecutionLeaseSummary is the bounded read-model projection of one Agent
// execution lease plus its FIFO context used by the Workflow Run UI.
type AgentExecutionLeaseSummary struct {
	Lease      AgentExecutionLease
	AheadCount int64
	QueueCount int64
}

// EffectQueueContext is the read-model projection that lets the Workflow Run UI
// show a callback effect's frozen responder, queue phase/position, and the
// current execution-lease owner without per-effect N+1 scans.
type EffectQueueContext struct {
	EffectID                 string
	CallbackResponderAgentID string
	QueueStatus              string
	EnqueueSeq               int64
	AheadCount               int64
	OwnerRunID               string
	OwnerActivationID        string
	OwnerEffectID            string
	OwnerNodeID              string
	HasLease                 bool
}

// LoadEffectQueueContexts returns the queue/lease context for every effect of
// one Run in a single bounded, workspace-scoped query. The held execution lease
// for a (workspace, responder) is joined by responder (not by effect id), so a
// queued effect also sees who currently owns the Agent capacity ahead of it.
// IsLeaseOwner is a separate comparison of the owner effect against this effect.
func (store *Store) LoadEffectQueueContexts(ctx context.Context, workspaceID, runID string) (map[string]EffectQueueContext, error) {
	if err := store.Validate(); err != nil {
		return nil, err
	}
	workspaceID, runID = strings.TrimSpace(workspaceID), strings.TrimSpace(runID)
	if workspaceID == "" || runID == "" {
		return nil, storeerr.BadRequest("workspaceId and runId are required")
	}
	rows, err := store.db.Query(ctx, `
SELECT e.id,
       COALESCE(e.callback_responder_agent_id, ''),
       COALESCE(q.status, ''),
       COALESCE(q.enqueue_seq, 0),
       COALESCE((
           SELECT COUNT(*)
           FROM workflow_agent_callback_queue q2
           WHERE q2.workspace_id = e.workspace_id
             AND q2.responder_agent_id = e.callback_responder_agent_id
             AND q2.status IN ('queued','dispatching')
             AND q2.enqueue_seq < q.enqueue_seq
       ), 0) AS ahead_count,
       COALESCE(owner.run_id, ''),
       COALESCE(owner.activation_id, ''),
       COALESCE(owner.effect_id, ''),
       COALESCE(owner.node_id, ''),
       CASE WHEN owner.status = 'held' AND owner.effect_id = e.id THEN 1 ELSE 0 END AS is_lease_owner
FROM workflow_run_effects e
LEFT JOIN workflow_agent_callback_queue q
  ON q.workspace_id = e.workspace_id AND q.effect_id = e.id
LEFT JOIN workflow_agent_execution_leases owner
  ON owner.workspace_id = e.workspace_id
 AND owner.responder_agent_id = e.callback_responder_agent_id
 AND owner.status = 'held'
WHERE e.workspace_id = $1 AND e.run_id = $2`, workspaceID, runID)
	if err != nil {
		return nil, fmt.Errorf("load workflow Run effect queue contexts: %w", err)
	}
	defer rows.Close()
	contexts := make(map[string]EffectQueueContext)
	for rows.Next() {
		var value EffectQueueContext
		var isLeaseOwner int
		if err := rows.Scan(&value.EffectID, &value.CallbackResponderAgentID, &value.QueueStatus, &value.EnqueueSeq,
			&value.AheadCount, &value.OwnerRunID, &value.OwnerActivationID, &value.OwnerEffectID, &value.OwnerNodeID, &isLeaseOwner); err != nil {
			return nil, fmt.Errorf("scan workflow Run effect queue context: %w", err)
		}
		value.HasLease = isLeaseOwner == 1
		contexts[value.EffectID] = value
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("iterate workflow Run effect queue contexts: %w", err)
	}
	return contexts, nil
}

// ReconcileAgentExecutionLeases releases held execution leases whose owning
// Run or callback effect is provably terminal (succeeded/failed/canceled) or
// whose owner effect row is gone, and terminalizes their queue items, so the
// next FIFO head can run. It NEVER releases a live owner: awaiting, pending
// retry, and dispatching/reclaiming callbacks are preserved regardless of age.
//
// The scan is workspace-group scoped, index-driven, and bounded; each lease is
// locked FOR UPDATE and its owner Run/Effect revalidated inside the same
// transaction under the placement -> Run -> Effect -> lease order so a
// concurrent claim/cancel cannot race it into a false release or a lock cycle.
// Returns the number of leases released.
func (store *Store) ReconcileAgentExecutionLeases(ctx context.Context, workspaceGroupID string, limit int) (int, error) {
	if err := store.Validate(); err != nil {
		return 0, err
	}
	bounded, err := boundedLimit(limit)
	if err != nil {
		return 0, err
	}
	if workspaceGroupID == "" {
		workspaceGroupID = store.workspaceGroupID
	}
	rows, err := store.db.Query(ctx, `
SELECT l.workspace_id, l.responder_agent_id, l.effect_id, l.run_id
FROM workflow_agent_execution_leases l
JOIN workspace_placements placement ON placement.workspace_id = l.workspace_id
WHERE l.status = 'held'
  AND placement.workspace_group_id = COALESCE(NULLIF($1,''), placement.workspace_group_id)
  AND placement.state IN ('active','draining','moving')
ORDER BY l.workspace_id, l.responder_agent_id, l.effect_id
LIMIT $2`, workspaceGroupID, bounded)
	if err != nil {
		return 0, fmt.Errorf("scan workflow Agent execution leases for reconcile: %w", err)
	}
	type leaseKey struct{ workspace, responder, effect, run string }
	var keys []leaseKey
	for rows.Next() {
		var key leaseKey
		if err := rows.Scan(&key.workspace, &key.responder, &key.effect, &key.run); err != nil {
			rows.Close()
			return 0, fmt.Errorf("scan workflow Agent execution lease: %w", err)
		}
		keys = append(keys, key)
	}
	rows.Close()
	if err := rows.Err(); err != nil {
		return 0, fmt.Errorf("iterate workflow Agent execution leases: %w", err)
	}
	released := 0
	for _, key := range keys {
		releasedOne, err := store.reconcileOneAgentExecutionLease(ctx, key.workspace, key.responder, key.effect)
		if err != nil {
			return released, err
		}
		released += releasedOne
	}
	return released, nil
}

func (store *Store) reconcileOneAgentExecutionLease(ctx context.Context, workspaceID, responderAgentID, effectID string) (int, error) {
	var released int
	err := coredb.WithTx(ctx, store.db, func(tx coredb.Tx) error {
		if _, err := store.lockCandidatePlacement(ctx, tx, workspaceID); err != nil {
			return err
		}
		// Lock the owner Run first (if any), then the Effect, then the lease.
		err := tx.QueryRow(ctx, `
SELECT status FROM workflow_run_instances
WHERE workspace_id=$1 AND id=(SELECT run_id FROM workflow_agent_execution_leases WHERE workspace_id=$1 AND responder_agent_id=$2 AND effect_id=$3)
FOR UPDATE`, workspaceID, responderAgentID, effectID).Scan(new(string))
		if err != nil && !coredb.IsNotFound(err) {
			return fmt.Errorf("lock workflow Agent execution lease owner Run: %w", err)
		}
		effect, effectFound, err := loadEffectForReconcile(ctx, tx, workspaceID, effectID)
		if err != nil {
			return err
		}
		// A live effect is an active external-work barrier regardless of the
		// aggregate Run status. An awaiting callback (the Agent may reply any
		// time), a retryable pending receipt, and a dispatching/crossed-dispatch
		// effect all still own the execution lease and the FIFO head position.
		// Releasing them because the Run aggregate is terminal would let a
		// queued tail jump ahead while the callback is still in flight. The
		// effect row itself is the durable outbox authority: once it is terminal
		// (completed/failed/canceled) or missing (cascade-deleted, so no dispatch
		// can be tracked), no external work remains and the lease may be released
		// exactly once.
		if effectFound && !effectTerminalForReconcile(effect.Status) {
			return nil
		}
		// The lease release and the queue terminalization are one atomic
		// transaction, so a lease that is already 'released' here also has its
		// queue item already terminalized. Count only the reconciler whose UPDATE
		// actually flipped the row: concurrent reconcilers that scanned the same
		// held lease before the placement lock must not double-report a release.
		leaseResult, err := tx.Exec(ctx, `
UPDATE workflow_agent_execution_leases
SET status='released', updated_at=$4
WHERE workspace_id=$1 AND responder_agent_id=$2 AND effect_id=$3 AND status='held'`,
			workspaceID, responderAgentID, effectID, utc(time.Now()))
		if err != nil {
			return fmt.Errorf("reconcile release workflow Agent execution lease: %w", err)
		}
		if leaseResult.RowsAffected() == 0 {
			return nil
		}
		if _, err := tx.Exec(ctx, `
UPDATE workflow_agent_callback_queue
SET status='completed', updated_at=$3
WHERE workspace_id=$1 AND effect_id=$2 AND status IN ('queued','dispatching')`,
			workspaceID, effectID, utc(time.Now())); err != nil {
			return fmt.Errorf("reconcile terminalize workflow callback queue item: %w", err)
		}
		released = 1
		return nil
	})
	return released, err
}

// effectTerminalForReconcile reports whether an effect is provably terminal
// (the callback completed, failed, or was canceled) so its execution lease is
// safe to release. Every other status — awaiting, pending retry, delivered, and
// dispatching/crossed-dispatch — is a live external-work barrier and must keep
// the lease.
func effectTerminalForReconcile(status string) bool {
	switch status {
	case EffectStatusCompleted, EffectStatusFailed, EffectStatusCanceled:
		return true
	default:
		return false
	}
}

func loadEffectForReconcile(ctx context.Context, queryer coredb.QueryRower, workspaceID, effectID string) (Effect, bool, error) {
	effect, err := scanEffect(queryer.QueryRow(ctx, effectSelectSQL()+`
WHERE workspace_id=$1 AND id=$2`, workspaceID, effectID))
	if coredb.IsNotFound(err) {
		return Effect{}, false, nil
	}
	if err != nil {
		return Effect{}, false, fmt.Errorf("load workflow Agent execution lease owner effect: %w", err)
	}
	return effect, true, nil
}