package run

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

	"github.com/tidbcloud/AgentTeam/internal/core/domain"
	"github.com/tidbcloud/AgentTeam/internal/coredb"
	"github.com/tidbcloud/AgentTeam/internal/store/model"
	"github.com/tidbcloud/AgentTeam/internal/store/storeerr"
)

// Maximum number of automatic recovery sends for one awaiting callback. After
// the 5th send the effect stays awaiting but is never auto-woken again.
const maxRecoveryAttempts = domain.WorkflowRecoveryMaxAttempts

// recoveryScheduleMinutes is the per-attempt delay in minutes between recovery
// sends, anchored at the first non-working observation. Delays accumulate, so
// attempt k is sent at firstNonWorkingAt + (2^k - 1) minutes (1, 3, 7, 15, 31).
// The legacy effect attempt counter is monotonic recovery-send history. A
// successful Agent Workflow action starts a new epoch; only the independent
// recovery_cycle_* schedule projection is lazily rebased on its next
// authoritative tick.
var recoveryScheduleMinutes = []int{1, 2, 4, 8, 16}

// recoveryAttemptDelay returns the delay from firstNonWorkingAt for attempt k
// (1-indexed): the cumulative sum of recoveryScheduleMinutes[:k], which equals
// 2^k - 1 minutes.
func recoveryAttemptDelay(k int) time.Duration {
	if k < 1 {
		return 0
	}
	minutes := (1 << uint(k)) - 1
	return time.Duration(minutes) * time.Minute
}

type recoveryStepResult struct {
	working           bool
	send              bool
	attemptIndex      int
	firstNonWorkingAt *time.Time
	attemptCount      int
	nextRecoveryAt    *time.Time
}

// nextRecoveryStep is the pure decision for one RecoveryRunner tick of one
// awaiting callback. While the callback is working the persisted schedule is
// returned unchanged (no reset, no rewind, no timing). The first non-working
// observation anchors the 1/2/4/8/16 schedule; a send fires only when the
// effect is non-working, below the attempt cap, and past the scheduled next
// time.
func nextRecoveryStep(
	now time.Time,
	firstNonWorkingAt *time.Time,
	attemptCount int,
	nextRecoveryAt *time.Time,
	working bool,
) recoveryStepResult {
	if working {
		return recoveryStepResult{
			working: working, firstNonWorkingAt: firstNonWorkingAt,
			attemptCount: attemptCount, nextRecoveryAt: nextRecoveryAt,
		}
	}
	if firstNonWorkingAt == nil {
		anchored := utc(now)
		next := anchored.Add(recoveryAttemptDelay(1))
		return recoveryStepResult{
			firstNonWorkingAt: &anchored, attemptCount: attemptCount, nextRecoveryAt: &next,
		}
	}
	if attemptCount >= maxRecoveryAttempts {
		return recoveryStepResult{
			firstNonWorkingAt: firstNonWorkingAt, attemptCount: attemptCount, nextRecoveryAt: nextRecoveryAt,
		}
	}
	if nextRecoveryAt != nil && !now.Before(*nextRecoveryAt) {
		nextAttempt := attemptCount + 1
		next := firstNonWorkingAt.Add(recoveryAttemptDelay(nextAttempt + 1))
		return recoveryStepResult{
			send:              true,
			attemptIndex:      nextAttempt,
			firstNonWorkingAt: firstNonWorkingAt,
			attemptCount:      nextAttempt,
			nextRecoveryAt:    &next,
		}
	}
	return recoveryStepResult{
		firstNonWorkingAt: firstNonWorkingAt, attemptCount: attemptCount, nextRecoveryAt: nextRecoveryAt,
	}
}

// ListRecoveryCandidatesParams bounds the RecoveryRunner scan of awaiting
// send_message.callback effects that may be due for an automatic recovery
// observation or send.
type ListRecoveryCandidatesParams struct {
	WorkspaceGroupID string
	Now              time.Time
	AfterID          string
	Limit            int
}

// ListRecoveryCandidates returns awaiting send_message.callback effects that the
// RecoveryRunner must observe this pass. The scan keeps two disjoint families:
//
//  1. Effects in the durable park FSM (park_state IN ('arm_on_idle','parked'))
//     are always included while `recovery_paused_at IS NULL`, even before
//     next_recovery_at or after the attempt cap: the runner must keep observing
//     them so the arm_on_idle -> parked idle confirmation and the parked ->
//     working clear can transition.
//  2. Ordinary effects are included when not paused, have not exhausted the
//     current-cycle attempt cap, and either have never been observed
//     non-working (recovery_cycle_first_non_working_at IS NULL) or are past
//     their next current-cycle scheduled recovery.
//
// The RecoveryRunner still re-checks working (the canonical Agent runtime state
// via RuntimeWorkingReader, never the Workflow lease or callback FIFO) inside
// the atomic tick, so a freshly delivered callback that is currently working is
// skipped without starting any timing.
func (store *Store) ListRecoveryCandidates(ctx context.Context, params ListRecoveryCandidatesParams) ([]Effect, error) {
	if err := store.Validate(); err != nil {
		return nil, err
	}
	params.WorkspaceGroupID, params.AfterID = strings.TrimSpace(params.WorkspaceGroupID), strings.TrimSpace(params.AfterID)
	groupID := store.workspaceGroupID
	if groupID == "" {
		groupID = params.WorkspaceGroupID
	} else if params.WorkspaceGroupID != "" && params.WorkspaceGroupID != groupID {
		return nil, storeerr.BadRequest("recovery workspace group differs from store routing scope")
	}
	limit, err := boundedLimit(params.Limit)
	if err != nil {
		return nil, err
	}
	if params.Now.IsZero() {
		params.Now = time.Now()
	}
	params.Now = utc(params.Now)
	rows, err := store.db.Query(ctx, `SELECT effect.* FROM (`+effectSelectSQL()+`) effect
JOIN workflow_run_instances run
  ON run.workspace_id=effect.workspace_id AND run.id=effect.run_id
JOIN workspace_placements placement ON placement.workspace_id=effect.workspace_id
LEFT JOIN workflow_agent_recovery_states recovery_state
  ON recovery_state.workspace_id=effect.workspace_id
 AND recovery_state.agent_id=effect.callback_responder_agent_id
WHERE effect.status='awaiting'
  AND effect.effect_kind='send_message'
  AND effect.callback_responder_agent_id IS NOT NULL
  AND effect.callback_responder_agent_id <> ''
  AND effect.recovery_paused_at IS NULL
  AND (
    effect.park_state IN ('arm_on_idle','parked')
    OR effect.recovery_epoch <> COALESCE(recovery_state.recovery_epoch, 0)
    OR (
      COALESCE(recovery_state.recovery_count, 0) < $2
      AND effect.recovery_cycle_attempt_count < $2
      AND (effect.recovery_cycle_first_non_working_at IS NULL OR effect.recovery_cycle_next_recovery_at <= $3)
    )
  )
  AND effect.id > $1
  AND run.status IN ('running','waiting')
  AND placement.workspace_group_id=COALESCE(NULLIF($4,''),placement.workspace_group_id)
  AND placement.state IN ('active','draining','moving')
ORDER BY effect.id ASC LIMIT $5`,
		params.AfterID, maxRecoveryAttempts, params.Now, groupID, limit)
	if err != nil {
		return nil, fmt.Errorf("list workflow Run recovery candidates: %w", err)
	}
	defer rows.Close()
	values := make([]Effect, 0, limit)
	for rows.Next() {
		value, err := scanEffect(rows)
		if err != nil {
			return nil, fmt.Errorf("scan workflow Run recovery candidate: %w", err)
		}
		values = append(values, value)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("iterate workflow Run recovery candidates: %w", err)
	}
	return values, nil
}

// RecoveryTickParams identifies one awaiting callback for an automatic recovery
// observation or send. The Agent recovery state is locked by the tick and
// carries the shared cap/epoch. The caller never calls the tick for an
// unknown/offline state, which must cause zero transition/write.
//
// Observation is the immutable authoritative runtime observation sampled by the
// caller under a machine observation guard that prevents turn-state replacement
// while held. The caller keeps that guard valid until this row-locked
// transaction commits and releases it on every path. The store therefore trusts
// the observation as current: it NEVER re-reads runtime state and NEVER invokes
// any Hub or machine-resolution callback inside the transaction, so there is no
// DB->Hub or Hub->DB lock inversion inside the tick. A duplicate same-token
// observation is idempotent; unknown/offline is zero transition/write.
type RecoveryTickParams struct {
	WorkspaceID string
	RunID       string
	EffectID    string
	Now         time.Time
	// Observation is the immutable authoritative runtime observation fenced by
	// the caller's guard. The store uses its Working/Known judgment for the
	// conditional transition and performs a zero-write fail-safe on
	// unknown/offline observations.
	Observation RecoveryObservation
}

// RecoveryTickResult reports the durable recovery state transition of one tick.
// Send means the caller must render and deliver the current-cycle display
// attempt at DisplayAttemptIndex now. The durable Effect.RecoveryAttemptCount is the
// separate monotonic delivery sequence for message idempotency; callers must
// not derive a client message id from DisplayAttemptIndex. The caller (RecoveryRunner)
// owns the working-state determination via the real
// Agent runtime reader; both known working and known idle observations reach
// this one row-locked conditional transition so the park FSM and the ordinary
// recovery schedule share the same winner.
type RecoveryTickResult struct {
	Effect Effect
	Send   bool
	// DisplayAttemptIndex is the resettable, 1-based current-cycle display index used
	// only in the Recovery Prompt. It is intentionally not a delivery identity.
	DisplayAttemptIndex int
	FirstNonWorkingAt   *time.Time
	AttemptCount        int
	NextRecoveryAt      *time.Time
	// RecoveryCount and RecoveryEpoch are the Agent-scoped cycle values. The
	// AttemptCount/FirstNonWorkingAt/NextRecoveryAt fields above are the
	// resettable current-cycle schedule used by the runner; result.Effect keeps
	// the legacy monotonic RecoveryAttemptCount compatibility history.
	RecoveryCount int
	RecoveryEpoch int64
}

// RecoveryTick atomically applies one conditional transition to one awaiting
// callback for an authoritative known runtime observation (working or idle).
//
// Park FSM (separate durable state, never an overload of recovery_paused_at):
//   - arm_on_idle + idle -> parked with no recovery send (writes park_parked_at).
//   - arm_on_idle + working -> zero transition/write (stays armed).
//   - parked + working -> clears the park and atomically resets recovery_paused_at
//     and the current-cycle schedule; no recovery send fires. The legacy
//     recovery_attempt_count history is retained. A subsequent idle anchors a fresh ordinary
//     1/2/4/8/16-minute schedule.
//   - parked + idle -> zero transition/write (stays parked, no recovery send).
//
// Ordinary schedule (park_state == ""): a working observation leaves the
// current-cycle schedule untouched (no reset, no rewind, no timing); the first
// idle observation anchors recovery_cycle_first_non_working_at without sending, and a due send
// fires exactly once (the effect row is locked FOR UPDATE so concurrent ticks
// serialize and only one wins each slot). The tick never reads the Workflow
// capacity lease or callback FIFO for liveness, never touches runnableAt, and
// never releases the execution lease.
func (store *Store) RecoveryTick(ctx context.Context, params RecoveryTickParams) (RecoveryTickResult, error) {
	if err := store.Validate(); err != nil {
		return RecoveryTickResult{}, err
	}
	params.WorkspaceID, params.RunID, params.EffectID = strings.TrimSpace(params.WorkspaceID), strings.TrimSpace(params.RunID), strings.TrimSpace(params.EffectID)
	if params.WorkspaceID == "" || params.RunID == "" || params.EffectID == "" {
		return RecoveryTickResult{}, storeerr.BadRequest("workspaceId, runId, and effectId are required")
	}
	if params.Now.IsZero() {
		params.Now = time.Now()
	}
	params.Now = utc(params.Now)
	store.runRowLockAttempting("recovery_tick")
	var result RecoveryTickResult
	err := coredb.WithTx(ctx, store.db, func(tx coredb.Tx) error {
		effect, recoveryState, proceed, err := store.lockRecoveryTickEffect(ctx, tx, params)
		if err != nil || !proceed {
			return err
		}
		// Test-only deterministic fence: the effect/Run rows are locked FOR
		// UPDATE here. Inert in production (nil). The committed-DB full-chain test
		// holds this point while a Hub writer proves it cannot cross the
		// observation guard.
		if store.testBarriers.recoveryTick != nil {
			if err := store.testBarriers.recoveryTick(ctx); err != nil {
				return err
			}
		}
		if err := store.runRowLockBarrier("recovery_tick"); err != nil {
			return err
		}
		// Authoritative runtime-state fencing: the caller sampled an immutable
		// observation under a machine observation guard that prevents turn-state
		// replacement while held, and holds that guard until this transaction
		// commits. The store trusts the observation as current and never re-reads
		// runtime state or invokes any Hub/machine-resolution callback inside the
		// transaction. An unknown/offline observation is a zero transition/write
		// fail-safe.
		if !params.Observation.Known {
			result = RecoveryTickResult{
				Effect: effect, RecoveryCount: recoveryState.RecoveryCount,
				RecoveryEpoch: recoveryState.RecoveryEpoch,
			}
			return nil
		}
		working := params.Observation.Working
		result, err = applyRecoveryTickTransition(ctx, tx, params, effect, recoveryState, working)
		return err
	})
	if err != nil {
		return RecoveryTickResult{}, err
	}
	return result, nil
}

// lockRecoveryTickEffect acquires the candidate placement lock, locks the Run
// row, and locks the effect row FOR UPDATE. proceed=false means the Run is not
// in a recoverable status or the effect is paused/non-awaiting, so the caller
// must perform a zero transition/write.
func (store *Store) lockRecoveryTickEffect(ctx context.Context, tx coredb.Tx, params RecoveryTickParams) (Effect, AgentRecoveryState, bool, error) {
	if _, err := store.lockCandidatePlacement(ctx, tx, params.WorkspaceID); err != nil {
		return Effect{}, AgentRecoveryState{}, false, 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 Effect{}, AgentRecoveryState{}, false, nil
	case err != nil:
		return Effect{}, AgentRecoveryState{}, false, fmt.Errorf("lock workflow Run for recovery tick: %w", err)
	case runStatus != RunStatusRunning && runStatus != RunStatusWaiting:
		return Effect{}, AgentRecoveryState{}, false, 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 Effect{}, AgentRecoveryState{}, false, mapReadError("lock workflow Run effect for recovery tick", err)
	}
	if effect.Status != EffectStatusAwaiting || effect.RecoveryPausedAt != nil {
		// Paused callbacks are never scanned, but a paused effect that races
		// in keeps its durable pause: zero transition/write.
		return effect, AgentRecoveryState{}, false, nil
	}
	if !params.Observation.Known {
		// Unknown/offline observations are a strict no-write fail-safe. Do not
		// even lazily create the Agent recovery row on this path.
		return effect, AgentRecoveryState{}, true, nil
	}
	recoveryState, err := lockAgentRecoveryState(
		ctx, tx, params.WorkspaceID, effect.CallbackResponderAgentID, params.Now,
	)
	if err != nil {
		return Effect{}, AgentRecoveryState{}, false, err
	}
	return effect, recoveryState, true, nil
}

// applyRecoveryTickTransition applies the park FSM or the ordinary recovery
// schedule to the row-locked effect, given the authoritative observation
// judgment. All writes are conditional on the effect still being awaiting and
// unpaused.
func applyRecoveryTickTransition(ctx context.Context, tx coredb.Tx, params RecoveryTickParams, effect Effect, recoveryState AgentRecoveryState, working bool) (RecoveryTickResult, error) {
	baseResult := func(value Effect) RecoveryTickResult {
		return RecoveryTickResult{
			Effect: value, RecoveryCount: recoveryState.RecoveryCount, RecoveryEpoch: recoveryState.RecoveryEpoch,
		}
	}

	switch effect.ParkState {
	case EffectParkStateArmOnIdle:
		if working {
			// Authoritative working while armed: zero transition/write. The
			// FSM keeps waiting for the first authoritative idle to confirm
			// parked. An armed effect is always scanned independently of the
			// shared attempt cap.
			return baseResult(effect), nil
		}
		// First authoritative known idle: confirm parked with no recovery send.
		if _, err := tx.Exec(ctx, `
UPDATE workflow_run_effects
SET park_state='parked', park_parked_at=$4, recovery_epoch=$5, updated_at=$4
WHERE workspace_id=$1 AND run_id=$2 AND id=$3 AND status='awaiting' AND recovery_paused_at IS NULL AND park_state='arm_on_idle'`,
			params.WorkspaceID, params.RunID, params.EffectID, params.Now, recoveryState.RecoveryEpoch); err != nil {
			return RecoveryTickResult{}, fmt.Errorf("park workflow Run awaiting callback: %w", err)
		}
		effect.ParkState = EffectParkStateParked
		effect.ParkParkedAt = &params.Now
		effect.RecoveryEpoch = recoveryState.RecoveryEpoch
		return baseResult(effect), nil
	case EffectParkStateParked:
		if !working {
			// Authoritative idle while parked: zero transition/write, no
			// recovery send. Duplicate/stale idle observations are idempotent.
			return baseResult(effect), nil
		}
		// First authoritative known working after parked: clear the park and
		// atomically reset only the current-cycle schedule so a subsequent idle
		// anchors a fresh ordinary 1/2/4/8/16-minute schedule. The legacy
		// recovery-send history remains monotonic and observable.
		if _, err := tx.Exec(ctx, `
UPDATE workflow_run_effects
SET park_state='', park_cleared_at=$4,
    recovery_paused_at=NULL, recovery_cycle_attempt_count=0,
    recovery_cycle_first_non_working_at=NULL, recovery_cycle_next_recovery_at=NULL,
    recovery_epoch=$5, updated_at=$4
WHERE workspace_id=$1 AND run_id=$2 AND id=$3 AND status='awaiting' AND recovery_paused_at IS NULL AND park_state='parked'`,
			params.WorkspaceID, params.RunID, params.EffectID, params.Now, recoveryState.RecoveryEpoch); err != nil {
			return RecoveryTickResult{}, fmt.Errorf("clear parked workflow Run callback recovery: %w", err)
		}
		cleared := params.Now
		effect.ParkState = EffectParkStateNone
		effect.ParkClearedAt = &cleared
		effect.RecoveryCycleFirstNonWorkingAt = nil
		effect.RecoveryCycleNextRecoveryAt = nil
		effect.RecoveryCycleAttemptCount = 0
		effect.RecoveryPausedAt = nil
		effect.RecoveryEpoch = recoveryState.RecoveryEpoch
		return baseResult(effect), nil
	default:
		return applyOrdinaryRecoveryTickTransition(ctx, tx, params, effect, recoveryState, working)
	}
}

func applyOrdinaryRecoveryTickTransition(ctx context.Context, tx coredb.Tx, params RecoveryTickParams, effect Effect, recoveryState AgentRecoveryState, working bool) (RecoveryTickResult, error) {
	baseResult := func(value Effect) RecoveryTickResult {
		return RecoveryTickResult{
			Effect: value, RecoveryCount: recoveryState.RecoveryCount, RecoveryEpoch: recoveryState.RecoveryEpoch,
		}
	}
	// Rebase late observations from an older Agent recovery epoch before
	// evaluating the current-cycle schedule.
	staleEpoch := effect.RecoveryEpoch != recoveryState.RecoveryEpoch
	cycleEffect := effect
	if staleEpoch {
		cycleEffect.RecoveryCycleAttemptCount = 0
		cycleEffect.RecoveryCycleFirstNonWorkingAt = nil
		cycleEffect.RecoveryCycleNextRecoveryAt = nil
	}
	if staleEpoch && recoveryState.RecoveryCount >= maxRecoveryAttempts {
		if _, err := tx.Exec(ctx, `
UPDATE workflow_run_effects
SET recovery_cycle_attempt_count=0,
    recovery_cycle_first_non_working_at=NULL, recovery_cycle_next_recovery_at=NULL,
    recovery_epoch=$4, updated_at=$5
WHERE workspace_id=$1 AND run_id=$2 AND id=$3 AND status='awaiting' AND recovery_paused_at IS NULL`,
			params.WorkspaceID, params.RunID, params.EffectID, recoveryState.RecoveryEpoch, params.Now); err != nil {
			return RecoveryTickResult{}, fmt.Errorf("rebase capped workflow Run recovery state: %w", err)
		}
		effect.RecoveryCycleAttemptCount = 0
		effect.RecoveryCycleFirstNonWorkingAt = nil
		effect.RecoveryCycleNextRecoveryAt = nil
		effect.RecoveryEpoch = recoveryState.RecoveryEpoch
		return baseResult(effect), nil
	}

	// Ordinary recovery schedule. A working observation leaves an
	// in-progress same-epoch schedule untouched. A stale-cycle working
	// observation clears only the old current-cycle projection without
	// touching the legacy history or the shared count.
	step := nextRecoveryStep(params.Now, cycleEffect.RecoveryCycleFirstNonWorkingAt, cycleEffect.RecoveryCycleAttemptCount, cycleEffect.RecoveryCycleNextRecoveryAt, working)
	result := baseResult(effect)
	result.Send = step.send
	result.DisplayAttemptIndex = step.attemptIndex
	result.FirstNonWorkingAt = step.firstNonWorkingAt
	result.AttemptCount = step.attemptCount
	result.NextRecoveryAt = step.nextRecoveryAt
	if working && !staleEpoch {
		// zero transition/write: the runner observed working, so the
		// schedule must remain exactly as persisted.
		return result, nil
	}
	if working && staleEpoch {
		step = recoveryStepResult{}
		result.Send = false
		result.DisplayAttemptIndex = 0
		result.FirstNonWorkingAt = nil
		result.AttemptCount = 0
		result.NextRecoveryAt = nil
	}
	// The effect cap remains a compatibility guard, while the shared Agent
	// count is the authoritative reset/cap boundary. One transaction
	// advances both projections and the Agent count when a send wins.
	if step.send && (recoveryState.RecoveryCount >= maxRecoveryAttempts || cycleEffect.RecoveryCycleAttemptCount >= maxRecoveryAttempts) {
		result.Send = false
		result.DisplayAttemptIndex = 0
		result.AttemptCount = effect.RecoveryCycleAttemptCount
		result.FirstNonWorkingAt = effect.RecoveryCycleFirstNonWorkingAt
		result.NextRecoveryAt = effect.RecoveryCycleNextRecoveryAt
		return result, nil
	}
	newRecoveryCount := recoveryState.RecoveryCount
	legacyAttemptCount := effect.RecoveryAttemptCount
	if step.send {
		newRecoveryCount++
		if newRecoveryCount > maxRecoveryAttempts {
			return RecoveryTickResult{}, fmt.Errorf("Workflow Agent recovery count exceeded cap")
		}
		if legacyAttemptCount == int(^uint(0)>>1) {
			return RecoveryTickResult{}, fmt.Errorf("Workflow callback recovery history exhausted")
		}
		legacyAttemptCount++
		if _, err := tx.Exec(ctx, `
UPDATE workflow_agent_recovery_states
SET recovery_count=$3, updated_at=$4
WHERE workspace_id=$1 AND agent_id=$2 AND recovery_epoch=$5`,
			params.WorkspaceID, effect.CallbackResponderAgentID, newRecoveryCount, params.Now, recoveryState.RecoveryEpoch); err != nil {
			return RecoveryTickResult{}, fmt.Errorf("advance Workflow Agent recovery count: %w", err)
		}
		result.DisplayAttemptIndex = newRecoveryCount
	}
	if _, err := tx.Exec(ctx, `
UPDATE workflow_run_effects
SET recovery_cycle_first_non_working_at=$2,
    recovery_cycle_attempt_count=$3,
    recovery_cycle_next_recovery_at=$4,
    recovery_epoch=$5,
    updated_at=$6
WHERE workspace_id=$1 AND run_id=$7 AND id=$8 AND status='awaiting' AND recovery_paused_at IS NULL`,
		params.WorkspaceID, nullableTime(step.firstNonWorkingAt), step.attemptCount, nullableTime(step.nextRecoveryAt),
		recoveryState.RecoveryEpoch, params.Now, params.RunID, params.EffectID); err != nil {
		return RecoveryTickResult{}, fmt.Errorf("advance workflow Run recovery cycle state: %w", err)
	}
	// The legacy first/next fields are historical compatibility data, not a
	// resettable schedule. The only legacy field changed here is the public
	// attempt history, and only when a real recovery send wins.
	if step.send {
		if _, err := tx.Exec(ctx, `
UPDATE workflow_run_effects
SET recovery_attempt_count=$2
WHERE workspace_id=$1 AND run_id=$3 AND id=$4 AND status='awaiting' AND recovery_paused_at IS NULL`,
			params.WorkspaceID, legacyAttemptCount, params.RunID, params.EffectID); err != nil {
			return RecoveryTickResult{}, fmt.Errorf("advance workflow Run recovery history: %w", err)
		}
	}
	effect.RecoveryCycleFirstNonWorkingAt = step.firstNonWorkingAt
	effect.RecoveryCycleAttemptCount = step.attemptCount
	effect.RecoveryCycleNextRecoveryAt = step.nextRecoveryAt
	effect.RecoveryAttemptCount = legacyAttemptCount
	effect.RecoveryEpoch = recoveryState.RecoveryEpoch
	result.Effect = effect
	result.RecoveryCount = newRecoveryCount
	return result, nil
}

// CallbackRecoveryMutationParams carries the authenticated frozen callback
// principal into a row-locked pause/park mutation.
type CallbackRecoveryMutationParams struct {
	WorkspaceID    string
	RunID          string
	EffectID       string
	AgentID        string
	MemberID       string
	Now            time.Time
	IdempotencyKey string
}

// CallbackRecoveryActionResult reports one durable non-terminal callback
// mutation. Replayed is used only by the permanent pause receipt; park returns
// Applied for each convergent FSM cycle/duplicate and never reports a replay.
// Applied remains true for both a first park arm and an already-armed cycle.
type CallbackRecoveryActionResult struct {
	Applied  bool
	Replayed bool
}

// PauseAwaitingCallbackRecovery preserves the legacy store boundary used by
// recovery tests and internal callers that do not have a callback key. Public
// Agent callback transport uses the WithIdempotency variant below.
func (store *Store) PauseAwaitingCallbackRecovery(ctx context.Context, workspaceID, runID, effectID string, now time.Time) (bool, error) {
	result, err := store.PauseAwaitingCallbackRecoveryWithIdempotency(ctx, workspaceID, runID, effectID, now, "")
	return result.Applied, err
}

func (store *Store) PauseAwaitingCallbackRecoveryWithIdempotency(ctx context.Context, workspaceID, runID, effectID string, now time.Time, idempotencyKey string) (CallbackRecoveryActionResult, error) {
	return store.mutateAwaitingCallbackRecovery(ctx, workspaceID, runID, effectID, now, "pause", idempotencyKey)
}

func (store *Store) PauseAwaitingCallbackRecoveryForAgent(ctx context.Context, params CallbackRecoveryMutationParams) (CallbackRecoveryActionResult, error) {
	return store.mutateAwaitingCallbackRecoveryForAgent(ctx, params, "pause")
}

// ParkAwaitingCallbackRecovery preserves the legacy store boundary. Public
// Agent callback transport may use the compatibility WithIdempotency variant,
// but park ignores that key and never writes a lifecycle receipt.
func (store *Store) ParkAwaitingCallbackRecovery(ctx context.Context, workspaceID, runID, effectID string, now time.Time) (bool, error) {
	result, err := store.ParkAwaitingCallbackRecoveryWithIdempotency(ctx, workspaceID, runID, effectID, now, "")
	return result.Applied, err
}

func (store *Store) ParkAwaitingCallbackRecoveryWithIdempotency(ctx context.Context, workspaceID, runID, effectID string, now time.Time, idempotencyKey string) (CallbackRecoveryActionResult, error) {
	return store.mutateAwaitingCallbackRecovery(ctx, workspaceID, runID, effectID, now, "park", idempotencyKey)
}

func (store *Store) ParkAwaitingCallbackRecoveryForAgent(ctx context.Context, params CallbackRecoveryMutationParams) (CallbackRecoveryActionResult, error) {
	return store.mutateAwaitingCallbackRecoveryForAgent(ctx, params, "park")
}

func (store *Store) mutateAwaitingCallbackRecovery(ctx context.Context, workspaceID, runID, effectID string, now time.Time, action, idempotencyKey string) (CallbackRecoveryActionResult, error) {
	return store.mutateAwaitingCallbackRecoveryForAgent(ctx, CallbackRecoveryMutationParams{
		WorkspaceID: workspaceID, RunID: runID, EffectID: effectID, Now: now, IdempotencyKey: idempotencyKey,
	}, action)
}

func (store *Store) mutateAwaitingCallbackRecoveryForAgent(ctx context.Context, params CallbackRecoveryMutationParams, action string) (CallbackRecoveryActionResult, error) {
	workspaceID, runID, effectID := params.WorkspaceID, params.RunID, params.EffectID
	now, idempotencyKey := params.Now, params.IdempotencyKey
	if err := store.Validate(); err != nil {
		return CallbackRecoveryActionResult{}, err
	}
	workspaceID, runID, effectID = strings.TrimSpace(workspaceID), strings.TrimSpace(runID), strings.TrimSpace(effectID)
	idempotencyKey = strings.TrimSpace(idempotencyKey)
	if workspaceID == "" || runID == "" || effectID == "" {
		return CallbackRecoveryActionResult{}, storeerr.BadRequest("workspaceId, runId, and effectId are required")
	}
	if action != "pause" && action != "park" {
		return CallbackRecoveryActionResult{}, storeerr.BadRequest("callback recovery action is invalid")
	}
	// Park is a repeatable FSM transition, not a lifecycle-wide receipt. Keep
	// accepting a legacy client key at the boundary, but never use it for
	// replay/conflict semantics or persist it in result_json.
	if action == "park" {
		idempotencyKey = ""
	}
	if idempotencyKey != "" && len(idempotencyKey) > 128 {
		return CallbackRecoveryActionResult{}, storeerr.BadRequest("callback idempotency key is too long")
	}
	if now.IsZero() {
		now = time.Now()
	}
	now = utc(now)
	store.runRowLockAttempting(action)
	var outcome CallbackRecoveryActionResult
	err := coredb.WithTx(ctx, store.db, func(tx coredb.Tx) error {
		if _, err := store.lockCandidatePlacement(ctx, tx, 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`, workspaceID, runID).Scan(&runStatus)
		switch {
		case coredb.IsNotFound(err):
			return nil
		case err != nil:
			return fmt.Errorf("lock workflow Run for recovery %s: %w", action, err)
		case runStatus != RunStatusRunning && runStatus != RunStatusWaiting:
			return nil
		}
		if err := store.runRowLockBarrier(action); err != nil {
			return err
		}
		var status, parkState, resultJSON string
		var callbackResponderAgentID string
		var recoveryPausedAt *time.Time
		err = tx.QueryRow(ctx, `
SELECT status, park_state, recovery_paused_at, COALESCE(result_json, ''),
       COALESCE(callback_responder_agent_id, '')
FROM workflow_run_effects
WHERE workspace_id=$1 AND run_id=$2 AND id=$3
FOR UPDATE`, workspaceID, runID, effectID).Scan(&status, &parkState, &recoveryPausedAt, &resultJSON, &callbackResponderAgentID)
		if coredb.IsNotFound(err) {
			return nil
		}
		if err != nil {
			return fmt.Errorf("lock workflow Run callback recovery %s effect: %w", action, err)
		}
		if strings.TrimSpace(params.MemberID) != "" {
			if callbackResponderAgentID != strings.TrimSpace(params.AgentID) {
				return storeerr.NotFound("workflow Run callback not found")
			}
			if err := authorizeCallbackAgentQuery(ctx, tx, AgentPrincipal{
				WorkspaceID: workspaceID, AgentID: params.AgentID, MemberID: params.MemberID,
			}, workspaceID, runID, effectID); err != nil {
				return err
			}
		}
		existingAction, existingKey, parseErr := model.DecodeCallbackRecoveryActionReceipt([]byte(resultJSON))
		if parseErr != nil {
			return fmt.Errorf("decode workflow Run callback recovery %s receipt: %w", action, parseErr)
		}
		var legacyParkReceiptJSON []byte
		// Older releases persisted park as a callback-lifetime receipt. It is
		// intentionally ignored so that a later FSM cycle can arm again. When
		// that old row is re-armed, remove the compatibility fields so no new
		// lifecycle-wide park receipt remains durable.
		if existingAction == "park" {
			legacyParkReceiptJSON, err = model.ClearCallbackRecoveryActionReceipt([]byte(resultJSON))
			if err != nil {
				return err
			}
			existingAction, existingKey = "", ""
		}
		if action == "pause" && (existingAction != "" || existingKey != "") {
			if existingAction == action && existingKey == idempotencyKey && idempotencyKey != "" {
				outcome = CallbackRecoveryActionResult{Applied: true, Replayed: true}
				return nil
			}
			return storeerr.Conflict("workflow Run callback idempotency key conflicts with the committed non-terminal action")
		}
		if status != EffectStatusAwaiting {
			return nil
		}
		if action == "pause" && (recoveryPausedAt != nil || parkState != EffectParkStateNone) {
			return nil
		}
		if action == "park" {
			if recoveryPausedAt != nil {
				return nil
			}
			if parkState == EffectParkStateArmOnIdle || parkState == EffectParkStateParked {
				// Concurrent duplicate parks in one cycle converge on the existing
				// FSM state without creating a receipt or recovery side effect.
				outcome.Applied = true
				return nil
			}
		}
		var callbackResultJSON []byte
		if action == "pause" && idempotencyKey != "" {
			callbackResultJSON, err = model.WriteCallbackRecoveryActionReceipt([]byte(resultJSON), action, idempotencyKey)
			if err != nil {
				return err
			}
		}
		var query string
		var args []any
		if action == "pause" {
			query = `
UPDATE workflow_run_effects
SET recovery_paused_at=$4, next_recovery_at=NULL,
    recovery_cycle_next_recovery_at=NULL,
    park_state='', park_armed_at=NULL, park_parked_at=NULL, park_cleared_at=$4,
    result_json=COALESCE($5, result_json), updated_at=$4
WHERE workspace_id=$1 AND run_id=$2 AND id=$3
  AND status='awaiting' AND park_state=''`
			args = []any{workspaceID, runID, effectID, now, nullableJSON(callbackResultJSON)}
		} else {
			query = `
UPDATE workflow_run_effects
SET park_state='arm_on_idle', park_armed_at=$4,
    result_json=COALESCE($5, result_json), updated_at=$4
WHERE workspace_id=$1 AND run_id=$2 AND id=$3
  AND status='awaiting' AND recovery_paused_at IS NULL AND park_state=''`
			args = []any{workspaceID, runID, effectID, now, nullableJSON(legacyParkReceiptJSON)}
		}
		result, err := tx.Exec(ctx, query, args...)
		if err != nil {
			return fmt.Errorf("persist workflow Run callback recovery %s: %w", action, err)
		}
		outcome.Applied = result.RowsAffected() > 0
		return nil
	})
	if err != nil {
		return CallbackRecoveryActionResult{}, err
	}
	return outcome, nil
}