package run

import (
	"bytes"
	"context"
	"encoding/json"
	"fmt"
	"strings"
	"time"

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

func (store *Store) ListPendingEffects(ctx context.Context, params ListPendingEffectsParams) ([]EffectCandidate, 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("effect 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.id,effect.workspace_id,effect.run_id,
       placement.workspace_group_id,placement.state,placement.placement_epoch
FROM workflow_run_effects 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_execution_leases execution
  ON execution.workspace_id=effect.workspace_id
 AND execution.responder_agent_id=effect.callback_responder_agent_id
LEFT JOIN workflow_agent_callback_queue queue_item
  ON queue_item.workspace_id=effect.workspace_id AND queue_item.effect_id=effect.id
WHERE effect.id>$1
  AND (
    (effect.status='pending'
      AND (effect.retry_at IS NULL OR effect.retry_at <= $2)
      AND (effect.lease_token IS NULL OR effect.lease_expires_at <= $2))
    OR
    (effect.status='dispatching' AND effect.lease_expires_at <= $2)
    OR
    (effect.status='queued_for_agent'
      AND effect.callback_responder_agent_id IS NOT NULL
      AND effect.callback_responder_agent_id <> ''
      AND COALESCE(execution.status,'released')='released'
      AND queue_item.enqueue_seq IS NOT NULL
      AND NOT EXISTS (
        SELECT 1
        FROM workflow_agent_callback_queue earlier
        WHERE earlier.workspace_id=effect.workspace_id
          AND earlier.responder_agent_id=effect.callback_responder_agent_id
          AND earlier.status IN ('queued','dispatching')
          AND earlier.enqueue_seq < queue_item.enqueue_seq
      ))
  )
  AND run.status IN ('running','waiting')
  AND placement.workspace_group_id=COALESCE(NULLIF($3,''),placement.workspace_group_id)
  AND placement.state IN ('active','draining','moving')
ORDER BY effect.id ASC LIMIT $4`, params.AfterID, params.Now, groupID, limit)
	if err != nil {
		return nil, fmt.Errorf("list pending workflow Run effects: %w", err)
	}
	defer rows.Close()
	values := make([]EffectCandidate, 0, limit)
	for rows.Next() {
		var value EffectCandidate
		if err := rows.Scan(&value.EffectID, &value.WorkspaceID, &value.runID, &value.workspaceGroup, &value.placementState, &value.placementEpoch); err != nil {
			return nil, fmt.Errorf("scan pending workflow Run effect: %w", err)
		}
		values = append(values, value)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("iterate pending workflow Run effects: %w", err)
	}
	return values, nil
}

func (store *Store) ClaimEffect(ctx context.Context, params ClaimEffectParams) (Effect, bool, error) {
	if err := store.Validate(); err != nil {
		return Effect{}, false, err
	}
	params.WorkerID = strings.TrimSpace(params.WorkerID)
	candidate := params.Candidate
	if params.WorkerID == "" || len(params.WorkerID) > 128 || params.LeaseTTL <= 0 ||
		candidate.EffectID == "" || candidate.WorkspaceID == "" || candidate.runID == "" || candidate.workspaceGroup == "" ||
		!placementRoutable(candidate.placementState) || candidate.placementEpoch <= 0 {
		return Effect{}, false, storeerr.BadRequest("complete DB-issued effect candidate, bounded workerId, and positive leaseTTL are required")
	}
	if params.Now.IsZero() {
		params.Now = time.Now()
	}
	params.Now = utc(params.Now)
	token, err := identifiers.NewEntityID("wfl")
	if err != nil {
		return Effect{}, false, fmt.Errorf("generate workflow Run effect lease: %w", err)
	}
	expires := utc(params.Now.Add(params.LeaseTTL))
	var effect Effect
	var claimed bool
	err = coredb.WithTx(ctx, store.db, func(tx coredb.Tx) error {
		// Claims and cancellation share one explicit lock order. In particular,
		// do not use an UPDATE JOIN here: TiDB may choose an implicit Run/Effect
		// lock order that conflicts with cancellation's placement -> Run -> Effect
		// order.
		placement, err := store.lockCandidatePlacement(ctx, tx, candidate.WorkspaceID)
		if err != nil {
			return err
		}
		if placement.workspaceGroup != candidate.workspaceGroup || placement.state != candidate.placementState || placement.epoch != candidate.placementEpoch ||
			(store.workspaceGroupID != "" && store.workspaceGroupID != placement.workspaceGroup) {
			return nil
		}
		var runStatus string
		err = tx.QueryRow(ctx, `
SELECT status
FROM workflow_run_instances
WHERE workspace_id=$1 AND id=$2
FOR UPDATE`, candidate.WorkspaceID, candidate.runID).Scan(&runStatus)
		switch {
		case coredb.IsNotFound(err):
			return nil
		case err != nil:
			return fmt.Errorf("lock workflow Run for effect claim: %w", err)
		case runStatus != RunStatusRunning && runStatus != RunStatusWaiting:
			return nil
		}
		result, err := tx.Exec(ctx, `
UPDATE workflow_run_effects
SET status='dispatching',attempt_count=attempt_count+1,
    lease_owner=$4,lease_token=$5,lease_expires_at=$6,
    error_code=NULL,error_json=NULL,updated_at=$3
WHERE workspace_id=$1 AND run_id=$2 AND id=$7
  AND (
    (status='pending'
      AND (retry_at IS NULL OR retry_at <= $3)
      AND (lease_token IS NULL OR lease_expires_at <= $3))
    OR
    (status='dispatching' AND lease_expires_at <= $3)
    OR
    (status='queued_for_agent'
      AND callback_responder_agent_id IS NOT NULL
      AND callback_responder_agent_id <> ''
      AND NOT EXISTS (
        SELECT 1
        FROM workflow_agent_callback_queue earlier
        WHERE earlier.workspace_id = workflow_run_effects.workspace_id
          AND earlier.responder_agent_id = workflow_run_effects.callback_responder_agent_id
          AND earlier.status IN ('queued','dispatching')
          AND earlier.enqueue_seq < (
            SELECT this.enqueue_seq
            FROM workflow_agent_callback_queue this
            WHERE this.workspace_id = workflow_run_effects.workspace_id
              AND this.effect_id = workflow_run_effects.id
          )
      ))
  )`, candidate.WorkspaceID, candidate.runID, params.Now, params.WorkerID, token, expires, candidate.EffectID)
		if err != nil {
			return fmt.Errorf("claim workflow Run effect: %w", err)
		}
		if result.RowsAffected() == 0 {
			return nil
		}
		effect, err = scanEffect(tx.QueryRow(ctx, effectSelectSQL()+`
WHERE workspace_id=$1 AND run_id=$2 AND id=$3`, candidate.WorkspaceID, candidate.runID, candidate.EffectID))
		if err != nil {
			return fmt.Errorf("load claimed workflow Run effect: %w", err)
		}
		claimed = true
		return nil
	})
	return effect, claimed, err
}

// FreezeApprovalAssignee durably binds a rendered approval effect to the one
// member authorized to decide it. Only the exact currently dispatching effect
// lease token may perform the first write; the same member is idempotent and a
// different member conflicts. Workflow app callbacks must trust this frozen
// column rather than re-evaluating mutable authoring input.
func (store *Store) FreezeApprovalAssignee(ctx context.Context, workspaceID, runID, effectID, leaseToken, memberID string) error {
	if err := store.Validate(); err != nil {
		return err
	}
	workspaceID, runID, effectID = strings.TrimSpace(workspaceID), strings.TrimSpace(runID), strings.TrimSpace(effectID)
	leaseToken, memberID = strings.TrimSpace(leaseToken), strings.TrimSpace(memberID)
	if workspaceID == "" || runID == "" || effectID == "" || leaseToken == "" || memberID == "" || len(memberID) > 64 {
		return storeerr.BadRequest("workspaceId, runId, effectId, leaseToken, and bounded approval memberId are required")
	}
	now := utc(time.Now())
	return coredb.WithTx(ctx, store.db, func(tx coredb.Tx) error {
		effect, err := scanEffect(tx.QueryRow(ctx, effectSelectSQL()+`
WHERE workspace_id=$1 AND run_id=$2 AND id=$3 FOR UPDATE`, workspaceID, runID, effectID))
		if err != nil {
			return mapReadError("lock approval effect assignee", err)
		}
		if effect.Kind != EffectKindApproval || effect.Status != EffectStatusDispatching || effect.LeaseToken != leaseToken ||
			effect.LeaseExpiresAt == nil || !effect.LeaseExpiresAt.After(now) {
			return storeerr.Conflict("approval effect dispatch lease changed or expired")
		}
		if effect.ApprovalAssigneeMemberID != "" {
			if effect.ApprovalAssigneeMemberID == memberID {
				return nil
			}
			return storeerr.Conflict("approval effect assignee is already frozen")
		}
		result, err := tx.Exec(ctx, `
UPDATE workflow_run_effects
SET approval_assignee_member_id=$2,updated_at=$3
WHERE id=$1 AND effect_kind='request_approval' AND status='dispatching'
	  AND workspace_id=$4 AND run_id=$5 AND lease_token=$6 AND lease_expires_at>$3
	  AND approval_assignee_member_id IS NULL`, effectID, memberID, now, workspaceID, runID, leaseToken)
		if err != nil {
			return fmt.Errorf("freeze workflow approval assignee: %w", err)
		}
		if result.RowsAffected() == 0 {
			return storeerr.Conflict("approval effect dispatch lease changed or assignee was frozen")
		}
		return nil
	})
}

func (store *Store) ResolveEffect(ctx context.Context, params ResolveEffectParams) (Effect, []Event, error) {
	if err := store.Validate(); err != nil {
		return Effect{}, nil, err
	}
	params.WorkspaceID, params.RunID, params.EffectID, params.LeaseToken =
		strings.TrimSpace(params.WorkspaceID), strings.TrimSpace(params.RunID), strings.TrimSpace(params.EffectID), strings.TrimSpace(params.LeaseToken)
	params.Status, params.ErrorCode = strings.TrimSpace(params.Status), strings.TrimSpace(params.ErrorCode)
	params.CallbackResponderAgentID = strings.TrimSpace(params.CallbackResponderAgentID)
	if params.WorkspaceID == "" || params.RunID == "" || params.EffectID == "" || params.LeaseToken == "" ||
		!validEffectResolutionStatus(params.Status) || len(params.ErrorCode) > 128 || params.RunnableAt.IsZero() {
		return Effect{}, nil, storeerr.BadRequest("complete effect fence, valid resolution status, and runnableAt are required")
	}
	if params.Status == EffectStatusPending && params.RetryAt == nil {
		return Effect{}, nil, storeerr.BadRequest("pending effect resolution requires retryAt")
	}
	if params.Status != EffectStatusPending && params.RetryAt != nil {
		return Effect{}, nil, storeerr.BadRequest("only pending effect resolution may carry retryAt")
	}
	if params.Status != EffectStatusAwaiting && params.CallbackResponderAgentID != "" {
		return Effect{}, nil, storeerr.BadRequest("only awaiting effect resolution may freeze a callback responder")
	}
	if params.Now.IsZero() {
		params.Now = time.Now()
	}
	params.Now, params.RunnableAt = utc(params.Now), utc(params.RunnableAt)
	params.RetryAt, params.DeliveredAt, params.CompletedAt = utcPtr(params.RetryAt), utcPtr(params.DeliveredAt), utcPtr(params.CompletedAt)
	var err error
	params.ResultJSON, err = deterministicObject("workflow Run effect result", params.ResultJSON, false)
	if err != nil {
		return Effect{}, nil, storeerr.BadRequest(err.Error())
	}
	params.ErrorJSON, err = deterministicObject("workflow Run effect error", params.ErrorJSON, false)
	if err != nil {
		return Effect{}, nil, storeerr.BadRequest(err.Error())
	}
	var resolved Effect
	var events []Event
	err = coredb.WithTx(ctx, store.db, func(tx coredb.Tx) error {
		run, err := scanRun(tx.QueryRow(ctx, runSelectSQL()+` WHERE r.workspace_id=$1 AND r.id=$2 FOR UPDATE`, params.WorkspaceID, params.RunID))
		if err != nil {
			return mapReadError("lock workflow Run for effect resolution", err)
		}
		if run.Terminal() {
			return storeerr.Conflict("workflow Run is terminal; late effect result is fenced")
		}
		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", err)
		}
		if effect.Status != EffectStatusDispatching || effect.LeaseToken != params.LeaseToken {
			return storeerr.Conflict("workflow Run effect lease changed")
		}
		if effect.Kind == EffectKindMessage {
			port, receiptMessageID, err := workflowEffectResultMessageIdentity(params.ResultJSON)
			if err != nil {
				return storeerr.BadRequest("workflow Run message effect result has an invalid messageId")
			}
			requiresMessage := params.Status == EffectStatusAwaiting || (params.Status == EffectStatusCompleted && port == "completed")
			if requiresMessage && receiptMessageID == "" {
				return storeerr.Conflict("workflow Run message effect result is missing its bound message")
			}
			// Once the native message transaction has bound a transcript row, a
			// dispatcher response may still be lost before its receipt is written
			// here. Such an Effect must remain replayable: accepting a message-less
			// timeout/failure would make the durable Effect claim that no delivery
			// happened even though the visible contribution (and, for Task
			// callbacks, its watch handoff) already committed. Only the canonical
			// bound message receipt may terminalize that recovery window.
			terminal := params.Status == EffectStatusCompleted || params.Status == EffectStatusFailed
			if terminal && effect.MessageID != "" && receiptMessageID == "" {
				return storeerr.Conflict("workflow Run bound message effect cannot resolve without its message receipt")
			}
			// Any native writer receipt that names a visible message must already
			// have crossed the atomic transcript/binding transaction, independent
			// of the authored output port. Retry/failure receipts without a message
			// remain valid because no visible contribution was committed.
			if receiptMessageID != "" && effect.MessageID == "" {
				return storeerr.Conflict("workflow Run message effect result is missing its native message binding")
			}
			if receiptMessageID != "" && effect.MessageID != "" && receiptMessageID != effect.MessageID {
				return storeerr.Conflict("workflow Run message effect result does not match its bound message")
			}
			if params.Status == EffectStatusAwaiting && params.CallbackResponderAgentID == "" {
				return storeerr.BadRequest("awaiting workflow Run message callback requires a frozen responder")
			}
			if params.CallbackResponderAgentID != "" && effect.CallbackResponderAgentID != "" &&
				effect.CallbackResponderAgentID != params.CallbackResponderAgentID {

				// The responder was frozen before delivery. A replay must never
				// retarget the callback to a different Agent, so a receipt naming a
				// different responder is fenced at the durable boundary.
				return storeerr.Conflict("workflow Run callback responder changed after freeze")
			}
		} else if params.CallbackResponderAgentID != "" {
			return storeerr.BadRequest("only workflow Run message callbacks may freeze a responder")
		}
		// ResolveEffect needs the Effect row to address its backing Activation.
		// Although this locks Effect before Activation, the Run lock above
		// serializes every graph mutation and every competing receipt path, so a
		// child-level lock inversion cannot form.
		activation, err := scanActivation(tx.QueryRow(ctx, activationSelectSQL()+`
WHERE workspace_id=$1 AND run_id=$2 AND id=$3 FOR UPDATE`, params.WorkspaceID, params.RunID, effect.ActivationID))
		if err != nil {
			return mapReadError("lock workflow Run effect activation", err)
		}
		// A deterministic reducer failure may be draining an Effect whose
		// external write already crossed the dispatch boundary. The same drain
		// occurs when normal graph progress abandons the backing Activation (for
		// example join.any or a competing end.fail path). Once that Effect
		// reaches any definitive/non-retry state, wake the aggregate immediately
		// so it can cancel an awaiting receipt or finalize. Pending retries
		// deliberately retain their authored retry schedule so the idempotent
		// writer gets a chance to converge the durable receipt.
		runnableAt := params.RunnableAt
		if params.Status != EffectStatusPending && (run.ErrorCode != "" || activation.Status != ActivationStatusWaiting) {
			runnableAt = params.Now
		}
		result, err := tx.Exec(ctx, `
UPDATE workflow_run_effects
SET status=$5,result_json=$6,retry_at=$7,lease_owner=NULL,lease_token=NULL,lease_expires_at=NULL,
    error_code=$8,error_json=$9,delivered_at=$10,completed_at=$11,
    callback_responder_agent_id=COALESCE($12,callback_responder_agent_id),updated_at=$13,
    park_state=CASE WHEN $5 IN ('completed','failed') THEN '' ELSE park_state END,
    park_armed_at=CASE WHEN $5 IN ('completed','failed') THEN NULL ELSE park_armed_at END,
    park_parked_at=CASE WHEN $5 IN ('completed','failed') THEN NULL ELSE park_parked_at END,
    park_cleared_at=CASE WHEN $5 IN ('completed','failed') THEN $13 ELSE park_cleared_at END,
    recovery_paused_at=CASE WHEN $5 IN ('completed','failed') THEN NULL ELSE recovery_paused_at END,
    next_recovery_at=CASE WHEN $5 IN ('completed','failed') THEN NULL ELSE next_recovery_at END,
    recovery_cycle_next_recovery_at=CASE WHEN $5 IN ('completed','failed') THEN NULL ELSE recovery_cycle_next_recovery_at END
WHERE workspace_id=$1 AND run_id=$2 AND id=$3 AND lease_token=$4 AND status='dispatching'`,
			params.WorkspaceID, params.RunID, params.EffectID, params.LeaseToken, params.Status,
			nullableJSON(params.ResultJSON), nullableTime(params.RetryAt), nullableString(params.ErrorCode), nullableJSON(params.ErrorJSON),
			nullableTime(params.DeliveredAt), nullableTime(params.CompletedAt), nullableString(params.CallbackResponderAgentID), params.Now)
		if err != nil {
			return fmt.Errorf("resolve workflow Run effect: %w", err)
		}
		if result.RowsAffected() == 0 {
			return storeerr.Conflict("workflow Run effect lease changed")
		}
		// A terminal resolution releases the Agent execution lease and marks the
		// FIFO queue item terminal in the same transaction. Retryable pending and
		// awaiting keep the same owner/position and never double-deliver.
		if params.Status == EffectStatusCompleted || params.Status == EffectStatusFailed {
			if err := releaseCallbackAgentExecutionLease(ctx, tx, params.WorkspaceID, effect.CallbackResponderAgentID, params.EffectID, callbackQueueStatusCompleted, params.Now); err != nil {
				return err
			}
		}
		txStore := store.InTx(tx)
		var lastSeq int64
		events, lastSeq, err = txStore.insertEvents(ctx, run, params.Events, params.Now)
		if err != nil {
			return err
		}
		result, err = tx.Exec(ctx, `
UPDATE workflow_run_instances
SET runnable_at=$3,last_event_seq=$4,revision=revision+1,updated_at=$5
WHERE workspace_id=$1 AND id=$2 AND revision=$6`, params.WorkspaceID, params.RunID, runnableAt, lastSeq, params.Now, run.Revision)
		if err != nil {
			return fmt.Errorf("wake workflow Run after effect resolution: %w", err)
		}
		if result.RowsAffected() == 0 {
			return storeerr.Conflict("workflow Run changed during effect resolution")
		}
		resolved, err = scanEffect(tx.QueryRow(ctx, effectSelectSQL()+` WHERE workspace_id=$1 AND run_id=$2 AND id=$3`, params.WorkspaceID, params.RunID, params.EffectID))
		return err
	})
	return resolved, events, err
}

func validEffectResolutionStatus(status string) bool {
	switch status {
	case EffectStatusPending, EffectStatusDelivered, EffectStatusAwaiting, EffectStatusCompleted, EffectStatusFailed:
		return true
	default:
		return false
	}
}

func (store *Store) CompleteEffect(ctx context.Context, workspaceID, runID, effectID, leaseToken string, resultJSON []byte, events []EventDraft, now time.Time) (Effect, []Event, error) {
	if now.IsZero() {
		now = time.Now()
	}
	completed := utc(now)
	return store.ResolveEffect(ctx, ResolveEffectParams{
		WorkspaceID: workspaceID, RunID: runID, EffectID: effectID, LeaseToken: leaseToken,
		Status: EffectStatusCompleted, ResultJSON: resultJSON, CompletedAt: &completed,
		RunnableAt: completed, Events: events, Now: completed,
	})
}

// ResolveAwaitingEffect atomically accepts the first typed callback/decision,
// records its effect receipt, wakes the Run, and appends events. The reducer
// remains the only authority that may move the Activation: it consumes this
// receipt in a later fenced SaveTurn and updates state_json and the relational
// Activation together. An exact same-receipt retry is a replay; a different
// result or late timeout/cancellation loses the status CAS and is rejected.
func (store *Store) ResolveAwaitingEffect(ctx context.Context, params ResolveAwaitingEffectParams) (ResolveAwaitingEffectResult, error) {
	if err := store.Validate(); err != nil {
		return ResolveAwaitingEffectResult{}, err
	}
	normalized, err := normalizeResolveAwaitingEffectParams(params)
	if err != nil {
		return ResolveAwaitingEffectResult{}, err
	}
	op := "complete"
	if strings.TrimSpace(normalized.SelectedPort) == "failed" {
		op = "fail"
	}
	store.runRowLockAttempting(op)
	var resolved ResolveAwaitingEffectResult
	err = coredb.WithTx(ctx, store.db, func(tx coredb.Tx) error {
		resolved, err = store.resolveAwaitingEffectTx(ctx, tx, normalized)
		return err
	})
	return resolved, err
}

// normalizeResolveAwaitingEffectParams canonicalizes the identity and durable
// payloads of a callback completion before the row-locked transaction.
func normalizeResolveAwaitingEffectParams(params ResolveAwaitingEffectParams) (ResolveAwaitingEffectParams, error) {
	params.WorkspaceID = strings.TrimSpace(params.WorkspaceID)
	params.RunID = strings.TrimSpace(params.RunID)
	params.ActivationID = strings.TrimSpace(params.ActivationID)
	params.EffectID = strings.TrimSpace(params.EffectID)
	params.SelectedPort = strings.TrimSpace(params.SelectedPort)
	if params.WorkspaceID == "" || params.RunID == "" || params.ActivationID == "" || params.EffectID == "" ||
		params.SelectedPort == "" || len(params.SelectedPort) > 128 || params.RunnableAt.IsZero() {
		return ResolveAwaitingEffectParams{}, storeerr.BadRequest("complete awaiting effect identity, selectedPort, and runnableAt are required")
	}
	if params.Now.IsZero() {
		params.Now = time.Now()
	}
	params.Now, params.RunnableAt = utc(params.Now), utc(params.RunnableAt)
	var err error
	params.EffectResultJSON, err = deterministicObject("workflow Run awaiting effect result", params.EffectResultJSON, true)
	if err != nil {
		return ResolveAwaitingEffectParams{}, storeerr.BadRequest(err.Error())
	}
	params.ActivationOutputJSON, err = deterministicObject("workflow Run awaiting activation output", params.ActivationOutputJSON, true)
	if err != nil {
		return ResolveAwaitingEffectParams{}, storeerr.BadRequest(err.Error())
	}
	params.ScopeContextJSON, err = deterministicObject("workflow Run awaiting effect scope context", params.ScopeContextJSON, false)
	if err != nil {
		return ResolveAwaitingEffectParams{}, storeerr.BadRequest(err.Error())
	}
	return params, nil
}

// resolveAwaitingEffectTx performs the row-locked first-result-wins transition
// for one awaiting callback: it locks the placement (matching the documented
// placement -> Run -> Effect lock order shared by claims, cancellation, pause,
// park, and recovery tick), then the Run/activation/effect FOR UPDATE, validates
// the idempotency authority, completes the effect, releases the Agent execution
// lease, and wakes the Run with the events/revision bump.
func (store *Store) resolveAwaitingEffectTx(ctx context.Context, tx coredb.Tx, params ResolveAwaitingEffectParams) (ResolveAwaitingEffectResult, error) {
	var resolved ResolveAwaitingEffectResult
	// Complete/fail must share the placement->Run->Effect lock order so a
	// concurrent pause/park/cancel/tick cannot deadlock against the Run row held
	// here (TiDB).
	if _, err := store.lockCandidatePlacement(ctx, tx, params.WorkspaceID); err != nil {
		return ResolveAwaitingEffectResult{}, err
	}
	run, err := scanRun(tx.QueryRow(ctx, runSelectSQL()+` WHERE r.workspace_id=$1 AND r.id=$2 FOR UPDATE`, params.WorkspaceID, params.RunID))
	if err != nil {
		return ResolveAwaitingEffectResult{}, mapReadError("lock workflow Run for awaiting effect", err)
	}
	activation, err := scanActivation(tx.QueryRow(ctx, activationSelectSQL()+`
WHERE workspace_id=$1 AND run_id=$2 AND id=$3 FOR UPDATE`, params.WorkspaceID, params.RunID, params.ActivationID))
	if err != nil {
		return ResolveAwaitingEffectResult{}, mapReadError("lock workflow Run awaiting activation", err)
	}
	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 ResolveAwaitingEffectResult{}, mapReadError("lock workflow Run awaiting effect", err)
	}
	// Test-only deterministic fence at the row-locked boundary (inert in
	// production); the concurrent matrix pauses the first operation here.
	op := "complete"
	if strings.TrimSpace(params.SelectedPort) == "failed" {
		op = "fail"
	}
	if err := store.runRowLockBarrier(op); err != nil {
		return ResolveAwaitingEffectResult{}, err
	}
	if strings.TrimSpace(params.AgentID) != "" && strings.TrimSpace(params.MemberID) != "" {
		if err := authorizeCallbackAgentQuery(ctx, tx, AgentPrincipal{
			WorkspaceID: params.WorkspaceID, AgentID: params.AgentID, MemberID: params.MemberID,
		}, params.WorkspaceID, params.RunID, params.EffectID); err != nil {
			return ResolveAwaitingEffectResult{}, err
		}
	}
	replayed, err := validateAwaitingEffectResolution(run, activation, effect, params)
	if err != nil {
		return ResolveAwaitingEffectResult{}, err
	}
	if replayed {
		return ResolveAwaitingEffectResult{Run: run, Activation: activation, Effect: effect, Replayed: true}, nil
	}
	if params.PreCommitGuard != nil {
		if err := params.PreCommitGuard(ctx, tx, run, activation, effect); err != nil {
			return ResolveAwaitingEffectResult{}, err
		}
	}
	if err := applyResolveAwaitingEffectPreCommitMutation(ctx, tx, run, activation, effect, &params); err != nil {
		return ResolveAwaitingEffectResult{}, err
	}
	// The production pre-commit mutation has now attempted its real
	// transaction-local writes (for Workflow feedback this is the Human
	// message plus scope fanout outbox). The default-off fault seam is placed
	// here so an injected retryable error rolls back those writes together with
	// the effect receipt and card-side delivery ledger.
	if err := store.runPreCommitMutationFault(ctx, run, activation, effect); err != nil {
		return ResolveAwaitingEffectResult{}, err
	}
	result, err := tx.Exec(ctx, `
UPDATE workflow_run_effects
SET status='completed',result_json=$4,error_code=NULL,error_json=NULL,completed_at=$5,updated_at=$5,
    park_state='',park_armed_at=NULL,park_parked_at=NULL,park_cleared_at=$5,
    recovery_paused_at=NULL,next_recovery_at=NULL,
    recovery_cycle_next_recovery_at=NULL
WHERE workspace_id=$1 AND run_id=$2 AND id=$3 AND status='awaiting'`,
		params.WorkspaceID, params.RunID, params.EffectID, params.EffectResultJSON, params.Now)
	if err != nil {
		return ResolveAwaitingEffectResult{}, fmt.Errorf("complete workflow Run awaiting effect: %w", err)
	}
	if result.RowsAffected() == 0 {
		return ResolveAwaitingEffectResult{}, storeerr.Conflict("workflow Run awaiting effect changed")
	}
	// Completing the awaited callback releases the Agent execution lease and
	// marks the FIFO queue item terminal so the next queued callback promotes.
	if err := releaseCallbackAgentExecutionLease(ctx, tx, params.WorkspaceID, effect.CallbackResponderAgentID, params.EffectID, callbackQueueStatusCompleted, params.Now); err != nil {
		return ResolveAwaitingEffectResult{}, err
	}
	txStore := store.InTx(tx)
	events, lastSeq, err := txStore.insertEvents(ctx, run, params.Events, params.Now)
	if err != nil {
		return ResolveAwaitingEffectResult{}, err
	}
	result, err = tx.Exec(ctx, `
UPDATE workflow_run_instances
SET runnable_at=$3,last_event_seq=$4,revision=revision+1,updated_at=$5
WHERE workspace_id=$1 AND id=$2 AND revision=$6`,
		params.WorkspaceID, params.RunID, params.RunnableAt, lastSeq, params.Now, run.Revision)
	if err != nil {
		return ResolveAwaitingEffectResult{}, fmt.Errorf("wake workflow Run after awaiting effect: %w", err)
	}
	if result.RowsAffected() == 0 {
		return ResolveAwaitingEffectResult{}, storeerr.Conflict("workflow Run changed during awaiting effect completion")
	}
	resolved.Run, err = scanRun(tx.QueryRow(ctx, runSelectSQL()+` WHERE r.workspace_id=$1 AND r.id=$2`, params.WorkspaceID, params.RunID))
	if err != nil {
		return ResolveAwaitingEffectResult{}, fmt.Errorf("load workflow Run after awaiting effect completion: %w", err)
	}
	resolved.Activation, err = scanActivation(tx.QueryRow(ctx, activationSelectSQL()+` WHERE workspace_id=$1 AND run_id=$2 AND id=$3`, params.WorkspaceID, params.RunID, params.ActivationID))
	if err != nil {
		return ResolveAwaitingEffectResult{}, fmt.Errorf("load completed workflow Run activation: %w", err)
	}
	resolved.Effect, err = scanEffect(tx.QueryRow(ctx, effectSelectSQL()+` WHERE workspace_id=$1 AND run_id=$2 AND id=$3`, params.WorkspaceID, params.RunID, params.EffectID))
	if err != nil {
		return ResolveAwaitingEffectResult{}, fmt.Errorf("load completed workflow Run effect: %w", err)
	}
	resolved.Events = events
	return resolved, nil
}

func validateAwaitingEffectResolution(run Run, activation Activation, effect Effect, params ResolveAwaitingEffectParams) (bool, error) {
	if effect.ActivationID != activation.ID {
		return false, storeerr.Conflict("workflow Run effect belongs to a different activation")
	}
	if effect.Kind == EffectKindMessage {
		_, receiptMessageID, err := workflowEffectResultMessageIdentity(params.EffectResultJSON)
		if err != nil || receiptMessageID == "" {
			return false, storeerr.Conflict("workflow Run callback result has an invalid messageId")
		}
		if effect.MessageID == "" {
			return false, storeerr.Conflict("workflow Run callback result is missing its native message binding")
		}
		if receiptMessageID != effect.MessageID {
			return false, storeerr.Conflict("workflow Run callback result does not match its bound message")
		}
	}
	if effect.Status == EffectStatusCompleted {
		if !bytes.Equal(effect.ResultJSON, params.EffectResultJSON) {
			return false, storeerr.Conflict("workflow Run awaiting effect already completed with a different result")
		}
		return true, nil
	}
	if run.Terminal() || effect.Status != EffectStatusAwaiting || activation.Status != ActivationStatusWaiting {
		return false, storeerr.Conflict("workflow Run awaiting effect is no longer resolvable")
	}
	return false, nil
}

func applyResolveAwaitingEffectPreCommitMutation(
	ctx context.Context,
	tx coredb.Tx,
	run Run,
	activation Activation,
	effect Effect,
	params *ResolveAwaitingEffectParams,
) error {
	if params == nil || params.PreCommitMutation == nil {
		return nil
	}
	if err := params.PreCommitMutation(ctx, tx, run, activation, effect, &params.EffectResultJSON); err != nil {
		return err
	}
	canonical, err := mustNormalizedObject("workflow Run awaiting effect result", params.EffectResultJSON)
	if err != nil {
		return storeerr.BadRequest(err.Error())
	}
	params.EffectResultJSON = canonical
	return nil
}

func workflowEffectResultMessageIdentity(payload []byte) (string, string, error) {
	if len(bytes.TrimSpace(payload)) == 0 {
		return "", "", nil
	}
	var result struct {
		Port      string `json:"port"`
		MessageID string `json:"messageId"`
	}
	if err := json.Unmarshal(payload, &result); err != nil {
		return "", "", err
	}
	return strings.TrimSpace(result.Port), strings.TrimSpace(result.MessageID), nil
}