kandev/apps/backend/internal/task/service/service_workflow.go

package service

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"math"
	"sort"
	"time"

	"github.com/google/uuid"
	"go.uber.org/zap"

	v1 "github.com/kandev/kandev/pkg/api/v1"

	"github.com/kandev/kandev/internal/auth/authn"
	"github.com/kandev/kandev/internal/authz"
	"github.com/kandev/kandev/internal/common/constants"
	"github.com/kandev/kandev/internal/events"
	"github.com/kandev/kandev/internal/steptelemetry"
	"github.com/kandev/kandev/internal/task/models"
	"github.com/kandev/kandev/internal/task/repository/repoerrors"
	wfmodels "github.com/kandev/kandev/internal/workflow/models"
	workflowmove "github.com/kandev/kandev/internal/workflow/move"
)

const (
	workflowMoveFromStepIDKey = "from_step_id"
	workflowMoveIDKey         = "move_id"
	workflowMoveOptionsKey    = "options"
)

// ApproveSessionResult contains the result of approving a session
type ApproveSessionResult struct {
	Session      *models.TaskSession
	Task         *models.Task
	WorkflowStep *wfmodels.WorkflowStep
}

type primarySessionTaskStateRepository interface {
	UpdateTaskStateIfPrimarySessionState(
		context.Context,
		string,
		string,
		models.TaskSessionState,
		v1.TaskState,
	) (v1.TaskState, bool, error)
}

// ApproveSession approves a session's current step and moves it to the next step.
// It reads the step's on_turn_complete actions to determine where to transition.
// If no transition actions are configured, it falls back to the next step by position.
func (s *Service) ApproveSession(ctx context.Context, sessionID string) (*ApproveSessionResult, error) {
	result := &ApproveSessionResult{}

	session, err := s.sessions.GetTaskSession(ctx, sessionID)
	if err != nil {
		return nil, fmt.Errorf("failed to get session: %w", err)
	}
	// Approving advances the task's workflow step, so this must be owner-only.
	if err := s.authorizeTaskID(ctx, session.TaskID); err != nil {
		return nil, err
	}
	result.Session = session

	// Get the task to find its current workflow step
	task, err := s.tasks.GetTask(ctx, session.TaskID)
	if err != nil {
		return nil, fmt.Errorf("failed to get task: %w", err)
	}

	// Get the current workflow step to check for transition targets
	if task.WorkflowStepID != "" && s.workflowStepGetter != nil {
		step, err := s.workflowStepGetter.GetStep(ctx, task.WorkflowStepID)
		if err != nil {
			s.logger.Warn("failed to get workflow step for approval transition",
				zap.String("workflow_step_id", task.WorkflowStepID),
				zap.Error(err))
		} else if err := s.applyApprovalStepTransition(ctx, sessionID, step, result); err != nil {
			return nil, err
		}
	}

	if err := s.sessions.UpdateSessionReviewStatus(ctx, sessionID, "approved"); err != nil {
		return nil, fmt.Errorf("failed to update review status: %w", err)
	}
	if session, err := s.sessions.GetTaskSession(ctx, sessionID); err == nil {
		result.Session = session
	}

	return result, nil
}

// applyApprovalStepTransition resolves the next workflow step and updates session/task accordingly.
func (s *Service) applyApprovalStepTransition(ctx context.Context, sessionID string, step *wfmodels.WorkflowStep, result *ApproveSessionResult) error {
	newStepID := s.resolveApprovalNextStep(ctx, step)

	if newStepID == "" {
		s.logger.Info("session approved but no next step found (may be at final step)",
			zap.String("session_id", sessionID),
			zap.String("current_step", step.ID),
			zap.String("current_step_name", step.Name))
		return nil
	}

	moved, err := s.MoveTaskWithOptions(ctx, result.Session.TaskID, step.WorkflowID, newStepID, 0,
		MoveTaskOptions{
			StepHistoryTrigger:   wfmodels.StepTransitionTriggerApproval,
			StepHistorySessionID: sessionID,
			StepHistoryActor:     wfmodels.StepTransitionActorHuman,
		})
	if err != nil {
		return fmt.Errorf("failed to move task to next step after approval: %w", err)
	}
	result.Task = moved.Task
	result.WorkflowStep = moved.WorkflowStep

	// Reload session with new step
	result.Session, _ = s.sessions.GetTaskSession(ctx, sessionID)

	// Get the new workflow step for the response
	if result.WorkflowStep == nil {
		newStep, err := s.workflowStepGetter.GetStep(ctx, newStepID)
		if err == nil {
			result.WorkflowStep = newStep
		}
	}

	s.logger.Info("session approved and moved to next step",
		zap.String("session_id", sessionID),
		zap.String("from_step", step.ID),
		zap.String("to_step", newStepID))
	return nil
}

// resolveApprovalNextStep determines the target step ID from a step's on_turn_complete actions,
// falling back to the next step by position when no actions are configured.
func (s *Service) resolveApprovalNextStep(ctx context.Context, step *wfmodels.WorkflowStep) string {
	var newStepID string
	for _, action := range step.Events.OnTurnComplete {
		switch action.Type {
		case "move_to_next":
			nextStep, err := s.workflowStepGetter.GetNextStepByPosition(ctx, step.WorkflowID, step.Position)
			if err != nil {
				s.logger.Warn("failed to get next step by position",
					zap.String("workflow_id", step.WorkflowID),
					zap.Int("current_position", step.Position),
					zap.Error(err))
			} else if nextStep != nil {
				newStepID = nextStep.ID
			}
		case "move_to_step":
			if stepID, ok := action.Config["step_id"].(string); ok && stepID != "" {
				newStepID = stepID
			}
		}
		if newStepID != "" {
			return newStepID
		}
	}

	// Fall back to next step by position if no transition actions found
	if len(step.Events.OnTurnComplete) == 0 {
		nextStep, err := s.workflowStepGetter.GetNextStepByPosition(ctx, step.WorkflowID, step.Position)
		if err != nil {
			s.logger.Warn("failed to get next step by position for fallback",
				zap.String("workflow_id", step.WorkflowID),
				zap.Int("current_position", step.Position),
				zap.Error(err))
		} else if nextStep != nil {
			s.logger.Info("using next step by position for approval transition (fallback)",
				zap.String("current_step", step.Name),
				zap.String("next_step", nextStep.Name))
			newStepID = nextStep.ID
		}
	}

	return newStepID
}

// UpdateTaskState updates the state of a task, moves it to the matching column,
// and publishes a task.state_changed event
func (s *Service) UpdateTaskState(ctx context.Context, id string, state v1.TaskState) (*models.Task, error) {
	if err := s.authorizeTaskScope(ctx, id, authz.ScopeTaskWrite); err != nil {
		return nil, err
	}
	task, err := s.tasks.GetTask(ctx, id)
	if err != nil {
		return nil, err
	}

	oldState := task.State

	// Skip no-op state transitions to avoid duplicate events.
	if oldState == state {
		return task, nil
	}

	if err := s.tasks.UpdateTaskState(ctx, id, state); err != nil {
		s.logger.Error("failed to update task state", zap.String("task_id", id), zap.Error(err))
		return nil, err
	}

	// Reload task to get updated state
	task, err = s.tasks.GetTask(ctx, id)
	if err != nil {
		return nil, err
	}

	s.logger.Info("task state updated",
		zap.String("task_id", id),
		zap.String("workflow_step_id", task.WorkflowStepID),
		zap.String("state", string(task.State)))

	s.publishTaskEvent(ctx, events.TaskStateChanged, task, &oldState)
	s.logger.Info("task state changed",
		zap.String("task_id", id),
		zap.String("old_state", string(oldState)),
		zap.String("new_state", string(state)))

	return task, nil
}

// UpdateTaskStateIfCurrentIn transitions state only when the task is currently
// in one of the allowed values. Publishes task.state_changed only when a row
// changes.
func (s *Service) UpdateTaskStateIfCurrentIn(
	ctx context.Context, id string, state v1.TaskState, allowed []v1.TaskState,
) (bool, error) {
	oldState, updated, err := s.tasks.UpdateTaskStateIfCurrentIn(ctx, id, state, allowed)
	if err != nil || !updated {
		return false, err
	}
	// Unreachable for current callers (allowed never includes state) — kept so a
	// future caller that includes state in allowed still skips a duplicate publish.
	if oldState == state {
		return false, nil
	}

	task, err := s.tasks.GetTask(ctx, id)
	if err != nil {
		return true, err
	}
	// The CAS wrote `state`; pin it on the payload so a concurrent transition
	// between commit and read cannot publish a mismatched new_state.
	task.State = state

	s.logger.Info("task state updated",
		zap.String("task_id", id),
		zap.String("workflow_step_id", task.WorkflowStepID),
		zap.String("state", string(state)))

	s.publishTaskEvent(ctx, events.TaskStateChanged, task, &oldState)
	s.logger.Info("task state changed",
		zap.String("task_id", id),
		zap.String("old_state", string(oldState)),
		zap.String("new_state", string(state)))

	return true, nil
}

// UpdateTaskStateIfNotArchived is UpdateTaskStateIfCurrentIn without the
// prior-state constraint — for writers (IN_PROGRESS runtime reconciliation)
// that legitimately fire from many prior states and only need the
// archived-task freeze guarantee. Publishes task.state_changed only when a
// row changes.
func (s *Service) UpdateTaskStateIfNotArchived(
	ctx context.Context, id string, state v1.TaskState,
) (bool, error) {
	oldState, updated, err := s.tasks.UpdateTaskStateIfNotArchived(ctx, id, state)
	if err != nil || !updated {
		return false, err
	}
	if oldState == state {
		return false, nil
	}

	task, err := s.tasks.GetTask(ctx, id)
	if err != nil {
		return true, err
	}
	// The CAS wrote `state`; pin it on the payload so a concurrent transition
	// between commit and read cannot publish a mismatched new_state.
	task.State = state

	s.logger.Info("task state updated",
		zap.String("task_id", id),
		zap.String("workflow_step_id", task.WorkflowStepID),
		zap.String("state", string(state)))

	s.publishTaskEvent(ctx, events.TaskStateChanged, task, &oldState)
	s.logger.Info("task state changed",
		zap.String("task_id", id),
		zap.String("old_state", string(oldState)),
		zap.String("new_state", string(state)))

	return true, nil
}

// UpdateTaskStateIfSessionState transitions task state only while its owning
// session remains in the expected state and the task remains unarchived.
// Publishes task.state_changed only when the guarded write changes state.
func (s *Service) UpdateTaskStateIfSessionState(
	ctx context.Context,
	taskID, sessionID string,
	expectedSessionState models.TaskSessionState,
	state v1.TaskState,
) (bool, error) {
	return s.updateTaskStateIfSessionState(
		ctx, taskID, sessionID, expectedSessionState, state, false,
	)
}

// UpdateTaskStateIfPrimarySessionState also requires the named session to
// remain primary.
func (s *Service) UpdateTaskStateIfPrimarySessionState(
	ctx context.Context,
	taskID, sessionID string,
	expectedSessionState models.TaskSessionState,
	state v1.TaskState,
) (bool, error) {
	return s.updateTaskStateIfSessionState(
		ctx, taskID, sessionID, expectedSessionState, state, true,
	)
}

func (s *Service) updateTaskStateIfSessionState(
	ctx context.Context,
	taskID, sessionID string,
	expectedSessionState models.TaskSessionState,
	state v1.TaskState,
	requirePrimary bool,
) (bool, error) {
	var (
		oldState v1.TaskState
		updated  bool
		err      error
	)
	if requirePrimary {
		updater, ok := s.tasks.(primarySessionTaskStateRepository)
		if !ok {
			return false, errors.New("primary-session task state update is not supported")
		}
		oldState, updated, err = updater.UpdateTaskStateIfPrimarySessionState(
			ctx, taskID, sessionID, expectedSessionState, state,
		)
	} else {
		oldState, updated, err = s.tasks.UpdateTaskStateIfSessionState(
			ctx, taskID, sessionID, expectedSessionState, state,
		)
	}
	if err != nil || !updated {
		return false, err
	}
	if oldState == state {
		return false, nil
	}

	task, err := s.tasks.GetTask(ctx, taskID)
	if err != nil {
		return true, err
	}
	// Pin the state written by the guarded CAS so a later transition between
	// commit and reload cannot produce a mismatched event payload.
	task.State = state
	s.publishTaskEvent(ctx, events.TaskStateChanged, task, &oldState)
	return true, nil
}

// UpdateTaskMetadata updates ordinary task metadata while preserving
// server-managed deferred-launch and step-handoff records.
func (s *Service) UpdateTaskMetadata(ctx context.Context, id string, metadata map[string]interface{}) (*models.Task, error) {
	if err := s.authorizeTaskScope(ctx, id, authz.ScopeTaskWrite); err != nil {
		return nil, err
	}
	task, err := s.tasks.GetTask(ctx, id)
	if err != nil {
		return nil, err
	}

	// Merge metadata (existing keys are preserved, new keys are added/updated)
	if task.Metadata == nil {
		task.Metadata = make(map[string]interface{})
	}
	for k, v := range metadata {
		// Deferred launch ownership is server-managed. Preserve it even if a
		// future metadata endpoint forwards the whole request map here.
		if k == models.MetaKeyDeferredLaunch || k == models.MetaKeyStepHandoffCarry {
			continue
		}
		task.Metadata[k] = v
	}
	task.UpdatedAt = time.Now().UTC()

	if err := s.tasks.UpdateTaskPreservingDeferredLaunch(ctx, task); err != nil {
		s.logger.Error("failed to update task metadata", zap.String("task_id", id), zap.Error(err))
		return nil, err
	}

	// Reload rather than publish/return the pre-write snapshot: the
	// preserving update can win a concurrent ceiling CAS on deferred_launch,
	// and that field lives only in the database row from this point on.
	current, err := s.tasks.GetTask(ctx, id)
	if err != nil {
		return nil, err
	}

	s.PublishTaskUpdated(ctx, current)
	s.logger.Debug("task metadata updated", zap.String("task_id", id), zap.Any("metadata", metadata))
	return current, nil
}

// MoveTaskResult contains the result of a MoveTask operation.
type MoveTaskResult struct {
	Task         *models.Task
	WorkflowStep *wfmodels.WorkflowStep
	// FromStepID and Transitioned are read off Task's own write-transaction
	// result (Task.FromStepID / Task.WorkflowStepTransitionID != 0), not from
	// this call's earlier pre-move snapshot — see Task.FromStepID's doc.
	FromStepID   string
	Transitioned bool
	// WorkflowEntryIdentity identifies the committed workflow-step entry that
	// accepted this move. Empty when the write did not transition the task.
	WorkflowEntryIdentity string
	// MoveID correlates the one-shot entry options carried on the transient
	// move marker with the target-step entry. Empty for an option-less move.
	MoveID string
	// EntryOptions is the normalized one-shot override accepted for this move,
	// or nil for an ordinary move.
	EntryOptions *workflowmove.EntryOptions
}

// MoveTaskOptions controls non-default move behavior for trusted callers.
type MoveTaskOptions struct {
	AllowActivePrimarySession bool
	// AllowFailedToCompletedRecovery permits the trusted launch-recovery
	// action to complete a failed task when it moves into a validated terminal
	// workflow step. Ordinary task moves preserve failed and cancelled states.
	AllowFailedToCompletedRecovery bool
	// PreserveDeferredLaunch keeps the deferred launch intent when an internal
	// queue promotion changes workflow steps. Manual moves still clear it.
	PreserveDeferredLaunch bool
	// StepHistoryTrigger overrides the ADR 0015 audit-row trigger recorded for
	// this move. Zero value defaults to StepTransitionTriggerManual — callers
	// driving an approval-gated transition (ApproveSession) set
	// StepTransitionTriggerApproval instead.
	StepHistoryTrigger wfmodels.StepTransitionTrigger
	// StepHistorySessionID pins the ADR 0015 audit-row session_id to a
	// specific session, overriding the primary/active-session resolution
	// MoveTaskWithOptions otherwise uses. ApproveSession sets this to the
	// session it is actually approving — on a task with more than one
	// active session, resolvePrimaryOrActiveSession can pick a different
	// (primary) session than the one being approved.
	StepHistorySessionID string
	// StepHistoryActor identifies the caller. Agent moves must not inherit the
	// owner identity that MCP uses for authorization.
	StepHistoryActor wfmodels.StepTransitionActor
	// ExpectedWorkflowID guards a caller that resolved "the task's current
	// workflow" via a separate pre-read (rather than passing an explicit,
	// intentional target workflow) against a concurrent reassignment landing
	// between that read and this call. When set, MoveTaskWithOptions compares
	// it to the task's WorkflowID from the fresh GetTask this method already
	// performs and fails with ErrWorkflowResolutionConflict on a mismatch,
	// instead of silently reverting whatever the concurrent move just did.
	ExpectedWorkflowID *string
	// EntryOptions are one-shot values applied when the orchestrator enters the
	// target workflow step. They are persisted privately on a transient task
	// marker and are never included in task.moved event payloads.
	EntryOptions *workflowmove.EntryOptions
}

// ErrWorkflowResolutionConflict indicates a caller's pre-resolved "current
// workflow" (see MoveTaskOptions.ExpectedWorkflowID) no longer matches the
// task's actual workflow — a concurrent reassignment won the race. The move
// is rejected rather than silently reverting that reassignment.
//
// This is an alias for repoerrors.ErrWorkflowResolutionConflict, not a
// separate sentinel: the same check now also runs atomically inside the
// repository's write transaction (UpdateTaskIfWorkflowMatches,
// UpdateTaskWithWorkflowStepAdmissionAndState), which returns the repoerrors
// value directly. Keeping one sentinel means callers using errors.Is against
// either name catch both the pre-write fast-fail below and the write-time
// guard that actually closes the race.
var ErrWorkflowResolutionConflict = repoerrors.ErrWorkflowResolutionConflict

type workflowMoveLimitsRepository interface {
	CountTasksByWorkflowStepExcludingTask(ctx context.Context, stepID, excludeTaskID string) (int, error)
}

type workflowAdmittedCountRepository interface {
	CountAdmittedTasksByWorkflowStep(ctx context.Context, stepID string) (int, error)
}

type workflowLimitedMoveRepository interface {
	UpdateTaskIfWorkflowStepHasCapacity(ctx context.Context, task *models.Task, targetStepID, excludeTaskID string, limit int) error
}

type workflowMoveAdmissionRepository interface {
	UpdateTaskWithWorkflowStepAdmission(ctx context.Context, task *models.Task, sourceStepID, targetStepID string, limit int) (bool, error)
}

type workflowMoveAdmissionWithStateRepository interface {
	UpdateTaskWithWorkflowStepAdmissionAndState(
		ctx context.Context,
		task *models.Task,
		sourceStepID string,
		targetStepID string,
		limit int,
		admittedState *v1.TaskState,
		queueExitPending bool,
		expectedWorkflowID string,
	) (bool, error)
}

// workflowMoveConflictRepository is the same-step counterpart of
// workflowMoveAdmissionWithStateRepository's expectedWorkflowID guard: when a
// move does not change workflow step, updateMovedTask writes through plain
// UpdateTask, which has no expected-workflow parameter (it is called from
// many unrelated, non-move sites). A caller that set
// MoveTaskOptions.ExpectedWorkflowID must instead route through this
// narrower method so the atomic in-transaction recheck still applies.
type workflowMoveConflictRepository interface {
	UpdateTaskIfWorkflowMatches(ctx context.Context, task *models.Task, expectedWorkflowID string) error
}

// metadataKeyRemoverRepository clears a single task metadata key in its own
// statement, without rewriting the concurrent fields a full UpdateTask would
// carry. Used to strip a stranded workflow_move_pending marker that a committed
// optioned move left behind when its write produced no step transition.
type metadataKeyRemoverRepository interface {
	RemoveTaskMetadataKey(ctx context.Context, taskID, key string) (bool, error)
}

type workflowQueuedTaskPromoter interface {
	PromoteQueuedTaskIfWorkflowStepHasCapacity(ctx context.Context, task *models.Task, fromStepID, destinationStepID string, limit int) (bool, error)
}

type workflowPullRepository interface {
	NextPullCandidateExcluding(ctx context.Context, stepID string, excludeTaskIDs []string) (*models.Task, error)
}

type workflowQueuedPullRepository interface {
	NextQueuedTaskForStepExcluding(ctx context.Context, feederStepID, destinationStepID string, excludeTaskIDs []string) (*models.Task, error)
}

const (
	priorityMedium = "medium"
	priorityLow    = "low"
)

// MoveTask moves a task to a different workflow step and position
func (s *Service) MoveTask(ctx context.Context, id string, workflowID string, workflowStepID string, position int) (*MoveTaskResult, error) {
	return s.MoveTaskWithOptions(ctx, id, workflowID, workflowStepID, position, MoveTaskOptions{})
}

// MoveTaskWithOptions moves a task with explicit caller options.
func (s *Service) MoveTaskWithOptions(
	ctx context.Context,
	id string,
	workflowID string,
	workflowStepID string,
	position int,
	opts MoveTaskOptions,
) (*MoveTaskResult, error) {
	if err := s.authorizeTaskScope(ctx, id, authz.ScopeTaskWrite); err != nil {
		return nil, err
	}
	task, err := s.tasks.GetTask(ctx, id)
	if err != nil {
		return nil, err
	}
	if opts.ExpectedWorkflowID != nil && task.WorkflowID != *opts.ExpectedWorkflowID {
		// Cheap fast-fail only: this GetTask is not inside a lock, so it
		// cannot by itself rule out a race landing between this read and the
		// eventual write below. That race is closed by the atomic,
		// in-transaction recheck updateMovedTask now performs at write time
		// (UpdateTaskIfWorkflowMatches / UpdateTaskWithWorkflowStepAdmissionAndState).
		// This early check only spares validateTaskMove and the session/state
		// lookups below when the mismatch is already visible from a plain read.
		return nil, fmt.Errorf("%w: resolved %q, task is now in %q",
			ErrWorkflowResolutionConflict, *opts.ExpectedWorkflowID, task.WorkflowID)
	}

	targetStep, err := s.validateTaskMove(ctx, task, workflowID, workflowStepID, opts)
	if err != nil {
		return nil, err
	}

	oldWorkflowID := task.WorkflowID
	oldStepID := task.WorkflowStepID
	oldState := task.State
	stepChanged := oldStepID != workflowStepID

	entryOptions, err := workflowmove.NormalizeEntryOptions(opts.EntryOptions, "")
	if err != nil {
		return nil, err
	}
	change := workflowmove.MoveChangePositionOnly
	if stepChanged {
		change = workflowmove.MoveChangeStep
	}
	if err := workflowmove.ValidateEntryOptions(entryOptions, change); err != nil {
		return nil, err
	}
	// Only a new optioned move conflicts with a marker already in flight. A
	// plain move falls through and clears a stranded marker below (moveID == "")
	// so a stale marker can never permanently block ordinary moves.
	if entryOptions != nil && stepChanged && task.Metadata != nil {
		if _, pending := task.Metadata[models.MetaKeyWorkflowMovePending]; pending {
			return nil, workflowmove.ErrMoveConflict
		}
	}
	// Entry options need a recipient. A step without auto-start can still
	// receive a one-shot hand-off through an active task session, but a
	// session-less move would otherwise accept and permanently drop the
	// instructions/profile after the task update is committed.
	if entryOptions != nil && targetStep != nil &&
		!targetStep.HasOnEnterAction(wfmodels.OnEnterAutoStartAgent) &&
		s.resolvePrimaryOrActiveSession(ctx, id) == nil {
		return nil, workflowmove.ErrEntryTargetUnavailable
	}
	moveID, optionsJSON, err := prepareWorkflowMoveMarker(entryOptions)
	if err != nil {
		return nil, err
	}

	if stepChanged && targetStep != nil && s.workflowMovePreflight != nil {
		currentSession := s.resolvePrimaryOrActiveSession(ctx, id)
		if err := s.workflowMovePreflight.PreflightWorkflowStepMove(ctx, id, currentSession, targetStep); err != nil {
			return nil, fmt.Errorf("failed to preflight workflow move: %w", err)
		}
	}
	stateAfterAdmission := *task
	if stepChanged {
		if err := s.syncTaskStateForWorkflowMove(ctx, &stateAfterAdmission, oldStepID, workflowStepID, opts); err != nil {
			return nil, fmt.Errorf("failed to sync task state for workflow move: %w", err)
		}
	}

	task.WorkflowID = workflowID
	task.WorkflowStepID = workflowStepID
	// A move naming the task's current step is not an arrival
	// (REQ-TASKS-KANBAN-TASK-REORDERING-001.28): it keeps the position it
	// already holds rather than the caller-supplied literal, which the
	// updateMovedTaskSameStep write path below leaves untouched. A step
	// change computes its own arrival position server-side further down
	// this call chain (updateTaskWithWorkflowStepAdmission), so the
	// caller-supplied position is ignored either way — position here is
	// never read again.
	if stepChanged {
		task.Position = position
		if task.Metadata == nil {
			task.Metadata = make(map[string]interface{})
		}
		task.WIPAdmitted = true
		task.QueuedForStepID = ""
		task.QueuedAt = nil
		task.Metadata[models.MetaKeyQueuedMoveExitPending] = map[string]interface{}{
			"from_step_id": oldStepID,
		}
		// The one-shot options ride on the pending marker itself: it is the
		// sole live transport for a direct optioned move. task.moved publishes
		// only the move ID; the orchestrator reads this marker at target entry.
		if moveID != "" {
			task.Metadata[models.MetaKeyWorkflowMovePending] = map[string]interface{}{
				workflowMoveFromStepIDKey: oldStepID,
				workflowMoveIDKey:         moveID,
				workflowMoveOptionsKey:    string(optionsJSON),
			}
		} else {
			delete(task.Metadata, models.MetaKeyWorkflowMovePending)
		}
		delete(task.Metadata, models.MetaKeyQueuedMoveExitCompleted)
		delete(task.Metadata, models.MetaKeyQueuePromotionPending)
		delete(task.Metadata, models.MetaKeyManualMoveLifecycleCompleted)
		if !opts.PreserveDeferredLaunch {
			models.DropWIPDeferredLaunch(task)
		}
	}
	task.UpdatedAt = time.Now().UTC()

	// Keep an admitted manual move's lifecycle barrier in the same task write as
	// the move whenever an active session exists. The admission repository removes
	// the queued-exit marker for admitted tasks, so this separate marker carries
	// the barrier across the task.moved event without changing WIP admission.
	sessionID := ""
	if stepChanged {
		if activeSession := s.resolvePrimaryOrActiveSession(ctx, id); activeSession != nil {
			sessionID = activeSession.ID
			task.Metadata[models.MetaKeyManualMoveLifecyclePending] = map[string]interface{}{
				"from_step_id": oldStepID,
			}
		}
	}

	var admittedState *v1.TaskState
	if stepChanged {
		admittedState = &stateAfterAdmission.State
	}
	if !stepChanged && opts.AllowFailedToCompletedRecovery && task.State == v1.TaskStateFailed {
		terminal, err := s.terminalWorkflowStep(ctx, workflowStepID)
		if err != nil {
			return nil, fmt.Errorf("failed to validate recovery target: %w", err)
		}
		if terminal {
			task.State = v1.TaskStateCompleted
		}
	}

	// manual_move only applies when no outer caller already declared a
	// trigger — an mcp_move set by the MCP handler must survive this inner
	// board-move default, since the agent (not a board click) is what caused
	// the move.
	moveCtx := ctx
	if !steptelemetry.HasTrigger(moveCtx) {
		actorKind, actorID := steptelemetry.HumanOrSystemActor(moveCtx)
		moveCtx = steptelemetry.WithAttribution(moveCtx, steptelemetry.Attribution{
			Trigger:   steptelemetry.TriggerManualMove,
			ActorKind: actorKind,
			ActorID:   actorID,
		})
	}

	_, err = s.updateMovedTask(moveCtx, task, oldStepID, targetStep, admittedState, opts)
	if err != nil {
		s.logger.Error("failed to move task", zap.String("task_id", id), zap.Error(err))
		return nil, err
	}

	// Captured now, before the post-commit GetTask refresh below replaces
	// task with a plain read that carries neither transient field (see
	// models.Task.FromStepID and FromWorkflowID) — these are the write
	// transaction's own result, not the pre-move snapshot computed above.
	resultFromWorkflowID := task.FromWorkflowID
	resultFromStepID := task.FromStepID
	resultTransitioned := task.WorkflowStepTransitionID != 0
	workflowEntryIdentity := ""
	if resultTransitioned && task.WorkflowStepTransitionID > 0 {
		workflowEntryIdentity = fmt.Sprintf("entry:%020d", task.WorkflowStepTransitionID)
	}
	if resultTransitioned && resultFromWorkflowID == "" {
		// Keep compatibility with repository implementations that predate the
		// transient source-workflow field. SQLite populates it from the write
		// transaction; this fallback preserves the previous event value for any
		// external test or adapter implementation that does not.
		resultFromWorkflowID = oldWorkflowID
	}

	// An optioned move (moveID != "") always changed step at read time —
	// ValidateEntryOptions rejects a position-only optioned move — so a
	// committed write with no transition means the task already occupied
	// workflowStepID when the atomic write ran: a concurrent move landed it
	// there between this call's read and its write. The one-shot options were
	// persisted on the pending marker for a target entry that will never fire
	// task.moved, so the marker would strand and the override silently never
	// apply. Clear it and report the conflict instead of returning a MoveID and
	// EntryOptions the orchestrator will never consume.
	if moveID != "" && !resultTransitioned {
		return s.rejectStrandedOptionedMove(ctx, task, resultFromWorkflowID)
	}

	s.publishTaskEvent(ctx, events.TaskUpdated, task, nil, resultFromWorkflowID)
	if oldState != task.State {
		s.publishTaskEvent(ctx, events.TaskStateChanged, task, &oldState)
	}

	// Publish task.moved event so the orchestrator can process on_exit/on_enter
	// actions. Gated on resultTransitioned (the write transaction's own
	// outcome), not the pre-write stepChanged: a concurrent same-workflow move
	// can land between this call's read and its write, so stepChanged/oldStepID
	// can name a step the task never actually left (stepChanged=true but the
	// commit was a no-op transition) or miss one it did leave (stepChanged=false
	// because this call's read already matched its own target, but the commit's
	// in-transaction read found a different actual "from" step). Every write
	// path reachable here (UpdateTask, UpdateTaskIfWorkflowMatches, and the
	// admission-with-state variant) sets WorkflowStepTransitionID, FromStepID,
	// and FromWorkflowID off the same in-transaction read, so the result fields
	// are always the source step and workflow the task actually left on this
	// commit.
	if resultTransitioned {
		s.publishTaskMovedEvent(ctx, task, resultFromWorkflowID, resultFromStepID, workflowStepID, sessionID, moveID)
		historySessionID := opts.StepHistorySessionID
		if historySessionID == "" {
			historySessionID = sessionID
		}
		s.recordManualStepTransition(ctx, historySessionID, resultFromStepID, workflowStepID, opts.StepHistoryTrigger, opts.StepHistoryActor)
		s.pullNextTaskOnVacate(ctx, resultFromStepID, task.ID)
		s.pullTasksFromNewFeederWork(ctx, workflowID, workflowStepID)
		refreshed, err := s.tasks.GetTask(ctx, task.ID)
		if err != nil {
			return nil, fmt.Errorf("failed to refresh task after feeder pull: %w", err)
		}
		if refreshed == nil {
			return nil, errors.New("failed to refresh task after feeder pull: repository returned nil task")
		}
		refreshed.Repositories = task.Repositories
		task = refreshed
	}

	s.logger.Info("task moved",
		zap.String("task_id", id),
		zap.String("workflow_id", workflowID),
		zap.String("workflow_step_id", workflowStepID),
		zap.Int("position", position))

	result := &MoveTaskResult{
		Task:                  task,
		FromStepID:            resultFromStepID,
		Transitioned:          resultTransitioned,
		WorkflowEntryIdentity: workflowEntryIdentity,
		MoveID:                moveID,
		EntryOptions:          entryOptions,
	}

	// Fetch the workflow step info if getter is available
	if s.workflowStepGetter != nil {
		step, err := s.workflowStepGetter.GetStep(ctx, task.WorkflowStepID)
		if err != nil {
			s.logger.Warn("failed to get workflow step for MoveTask response",
				zap.String("workflow_step_id", task.WorkflowStepID),
				zap.Error(err))
			// Don't fail the operation, just log and continue
		} else {
			result.WorkflowStep = step
		}
	}

	return result, nil
}

// rejectStrandedOptionedMove undoes an optioned move whose committed write
// produced no step transition (the task already occupied the target step). The
// write persisted the workflow_move_pending marker, but the zero-transition
// result means task.moved never fires to consume it, so the marker is removed
// with a single-key delete that does not clobber the concurrent move's fields.
// A TaskUpdated event keeps subscribers converged on the cleared row, and the
// caller receives ErrMoveConflict rather than a false success.
func (s *Service) rejectStrandedOptionedMove(ctx context.Context, task *models.Task, fromWorkflowID string) (*MoveTaskResult, error) {
	remover, ok := s.tasks.(metadataKeyRemoverRepository)
	if !ok {
		return nil, fmt.Errorf("metadata key remover repository unavailable for task %s", task.ID)
	}
	if _, err := remover.RemoveTaskMetadataKey(ctx, task.ID, models.MetaKeyWorkflowMovePending); err != nil {
		return nil, fmt.Errorf("clear stranded workflow move marker for task %s: %w", task.ID, err)
	}
	delete(task.Metadata, models.MetaKeyWorkflowMovePending)
	s.publishTaskEvent(ctx, events.TaskUpdated, task, nil, fromWorkflowID)
	return nil, workflowmove.ErrMoveConflict
}

// prepareWorkflowMoveMarker mints a correlation ID and encodes the one-shot
// options for the workflow_move_pending marker. The marker is the sole live
// transport for a direct optioned move; nil options mean an ordinary move.
func prepareWorkflowMoveMarker(entryOptions *workflowmove.EntryOptions) (string, json.RawMessage, error) {
	if entryOptions == nil {
		return "", nil, nil
	}
	optionsJSON, err := workflowmove.EncodeEntryOptionsJSON(entryOptions)
	if err != nil {
		return "", nil, fmt.Errorf("encode workflow move options: %w", err)
	}
	return uuid.NewString(), optionsJSON, nil
}

func (s *Service) terminalWorkflowStep(ctx context.Context, workflowStepID string) (bool, error) {
	if s.workflowStepGetter == nil || workflowStepID == "" {
		return false, nil
	}
	step, err := s.workflowStepGetter.GetStep(ctx, workflowStepID)
	if err != nil {
		return false, fmt.Errorf("failed to get workflow step %s: %w", workflowStepID, err)
	}
	if step == nil {
		return false, nil
	}
	nextStep, err := s.workflowStepGetter.GetNextStepByPosition(ctx, step.WorkflowID, step.Position)
	if err != nil {
		return false, fmt.Errorf("failed to get next workflow step after %s: %w", workflowStepID, err)
	}
	return wfmodels.IsTerminalStep(step, nextStep), nil
}

func (s *Service) syncTaskStateForWorkflowMove(ctx context.Context, task *models.Task, oldStepID, newStepID string, opts MoveTaskOptions) error {
	newTerminal, err := s.terminalWorkflowStep(ctx, newStepID)
	if err != nil {
		return err
	}
	if newTerminal {
		if task.State == v1.TaskStateFailed && opts.AllowFailedToCompletedRecovery {
			task.State = v1.TaskStateCompleted
		} else if !models.IsTerminalTaskState(task.State) {
			task.State = v1.TaskStateCompleted
		}
		return nil
	}
	if oldStepID == newStepID || task.State != v1.TaskStateCompleted {
		return nil
	}
	oldTerminal, err := s.terminalWorkflowStep(ctx, oldStepID)
	if err != nil {
		return err
	}
	if oldTerminal {
		task.State = v1.TaskStateTODO
	}
	return nil
}

// ReconcileVacatedStep fills available capacity in a workflow step from its
// same-step queue or configured feeder.
func (s *Service) ReconcileVacatedStep(ctx context.Context, vacatedStepID string) {
	s.pullNextTaskOnVacate(ctx, vacatedStepID, "")
}

func (s *Service) pullNextTaskOnVacate(ctx context.Context, vacatedStepID, excludeTaskID string) {
	// A queue/WIP reconciliation is always wip_pull, unconditionally
	// overriding whatever trigger the caller that vacated the step declared
	// — the vacating move and the resulting pull are two distinct ledger
	// rows with two distinct causes. No single session initiates a pull, so
	// actor kind is always system with no session.
	ctx = steptelemetry.WithAttribution(ctx, steptelemetry.Attribution{
		Trigger:   steptelemetry.TriggerWIPPull,
		ActorKind: steptelemetry.ActorSystem,
	})
	vacatedStep := s.reconcilableStep(ctx, vacatedStepID)
	if vacatedStep == nil {
		return
	}
	occupants, ok := s.currentAdmittedOccupants(ctx, vacatedStep.ID)
	if !ok || (vacatedStep.WIPLimit > 0 && occupants >= vacatedStep.WIPLimit) {
		return
	}
	skipped := map[string]struct{}{excludeTaskID: {}}
	for vacatedStep.WIPLimit <= 0 || occupants < vacatedStep.WIPLimit {
		pulled := s.promoteNextQueuedTask(ctx, vacatedStep, occupants, skipped)
		if !pulled {
			return
		}
		occupants++
	}
}

func (s *Service) reconcilableStep(ctx context.Context, vacatedStepID string) *wfmodels.WorkflowStep {
	if s.workflowStepGetter == nil || vacatedStepID == "" {
		return nil
	}
	vacatedStep, err := s.workflowStepGetter.GetStep(ctx, vacatedStepID)
	if err != nil || vacatedStep == nil {
		return nil
	}
	return vacatedStep
}

func (s *Service) currentAdmittedOccupants(ctx context.Context, stepID string) (int, bool) {
	limitsRepo, ok := s.tasks.(workflowAdmittedCountRepository)
	if !ok {
		fallback, fallbackOK := s.tasks.(workflowMoveLimitsRepository)
		if !fallbackOK {
			s.logger.Warn("cannot reconcile queued task: WIP count repository unavailable", zap.String("step_id", stepID))
			return 0, false
		}
		occupants, err := fallback.CountTasksByWorkflowStepExcludingTask(ctx, stepID, "")
		if err != nil {
			s.logger.Warn("cannot reconcile queued task: failed to count step", zap.String("step_id", stepID), zap.Error(err))
			return 0, false
		}
		return occupants, true
	}
	occupants, err := limitsRepo.CountAdmittedTasksByWorkflowStep(ctx, stepID)
	if err != nil {
		s.logger.Warn("cannot reconcile queued task: failed to count step",
			zap.String("step_id", stepID), zap.Error(err))
		return 0, false
	}
	return occupants, true
}

func (s *Service) promoteNextQueuedTask(ctx context.Context, targetStep *wfmodels.WorkflowStep, position int, skipped map[string]struct{}) bool {
	candidate, err := s.nextQueuedCandidate(ctx, targetStep, skipped)
	if err != nil || candidate == nil {
		return false
	}
	fromStepID := candidate.WorkflowStepID
	oldWorkflowID := candidate.WorkflowID
	promotionSessionID := ""
	if fromStepID != targetStep.ID {
		promotionSession, blocked := s.feederCandidateSession(ctx, candidate.ID)
		if blocked {
			skipped[candidate.ID] = struct{}{}
			return s.promoteNextQueuedTask(ctx, targetStep, position, skipped)
		}
		if promotionSession != nil {
			promotionSessionID = promotionSession.ID
		}
	}
	if moveLifecyclePending(candidate) {
		skipped[candidate.ID] = struct{}{}
		return s.promoteNextQueuedTask(ctx, targetStep, position, skipped)
	}
	if candidate.WorkflowStepID == targetStep.ID {
		return s.promoteSameStepQueuedTask(ctx, candidate, fromStepID, targetStep, position, skipped)
	}
	return s.promoteFeederQueuedTask(ctx, candidate, fromStepID, oldWorkflowID, targetStep, position, skipped, promotionSessionID)
}

func queuedMoveExitPending(task *models.Task) bool {
	if task == nil || task.Metadata == nil {
		return false
	}
	_, pending := task.Metadata[models.MetaKeyQueuedMoveExitPending]
	if !pending {
		return false
	}
	_, completed := task.Metadata[models.MetaKeyQueuedMoveExitCompleted]
	return !completed
}

func manualMoveLifecyclePending(task *models.Task) bool {
	if task == nil || task.Metadata == nil {
		return false
	}
	if _, pending := task.Metadata[models.MetaKeyManualMoveLifecyclePending]; !pending {
		return false
	}
	_, completed := task.Metadata[models.MetaKeyManualMoveLifecycleCompleted]
	return !completed
}

func moveLifecyclePending(task *models.Task) bool {
	return queuedMoveExitPending(task) || manualMoveLifecyclePending(task)
}

func (s *Service) promoteSameStepQueuedTask(ctx context.Context, candidate *models.Task, fromStepID string, targetStep *wfmodels.WorkflowStep, position int, skipped map[string]struct{}) bool {
	oldState := candidate.State
	if candidate.Metadata == nil {
		candidate.Metadata = make(map[string]interface{})
	}
	candidate.WIPAdmitted = true
	candidate.QueuedForStepID = ""
	candidate.QueuedAt = nil
	candidate.Position = position
	candidate.Metadata[models.MetaKeyQueuePromotionPending] = map[string]interface{}{
		"from_step_id": fromStepID,
	}
	if err := s.syncTaskStateForQueuePromotion(ctx, candidate, targetStep); err != nil {
		s.logger.Warn("failed to prepare same-step queued promotion", zap.String("task_id", candidate.ID), zap.Error(err))
		skipped[candidate.ID] = struct{}{}
		return s.promoteNextQueuedTask(ctx, targetStep, position, skipped)
	}
	supported, claimed, err := promoteQueuedTaskAtomically(ctx, s.tasks, candidate, fromStepID, targetStep.ID, targetStep.WIPLimit)
	if supported {
		return s.finishAtomicQueuedPromotion(ctx, candidate, targetStep, position, skipped, claimed, err, oldState)
	} else if admissionRepo, ok := s.tasks.(workflowMoveAdmissionRepository); ok {
		claimed, err := admissionRepo.UpdateTaskWithWorkflowStepAdmission(ctx, candidate, fromStepID, targetStep.ID, targetStep.WIPLimit)
		if err != nil {
			s.logger.Warn("failed to promote same-step queued task", zap.String("task_id", candidate.ID), zap.Error(err))
			skipped[candidate.ID] = struct{}{}
			return s.promoteNextQueuedTask(ctx, targetStep, position, skipped)
		}
		if !claimed {
			skipped[candidate.ID] = struct{}{}
			return s.promoteNextQueuedTask(ctx, targetStep, position, skipped)
		}
	} else if err := s.tasks.UpdateTask(ctx, candidate); err != nil {
		return false
	}
	s.publishTaskEvent(ctx, events.TaskUpdated, candidate, nil)
	if oldState != candidate.State {
		s.publishTaskEvent(ctx, events.TaskStateChanged, candidate, &oldState)
	}
	s.publishTaskEvent(ctx, events.TaskQueuePromoted, candidate, nil)
	return true
}

func (s *Service) finishAtomicQueuedPromotion(ctx context.Context, candidate *models.Task, targetStep *wfmodels.WorkflowStep, position int, skipped map[string]struct{}, claimed bool, err error, oldState v1.TaskState) bool {
	if err != nil {
		s.logger.Warn("failed to promote same-step queued task", zap.String("task_id", candidate.ID), zap.Error(err))
		return false
	}
	if !claimed {
		skipped[candidate.ID] = struct{}{}
		return s.promoteNextQueuedTask(ctx, targetStep, position, skipped)
	}
	s.publishTaskEvent(ctx, events.TaskUpdated, candidate, nil)
	if oldState != candidate.State {
		s.publishTaskEvent(ctx, events.TaskStateChanged, candidate, &oldState)
	}
	s.publishTaskEvent(ctx, events.TaskQueuePromoted, candidate, nil)
	return true
}

func promoteQueuedTaskAtomically(ctx context.Context, tasks interface{}, task *models.Task, fromStepID, destinationStepID string, limit int) (bool, bool, error) {
	promoter, ok := tasks.(workflowQueuedTaskPromoter)
	if !ok {
		return false, false, nil
	}
	claimed, err := promoter.PromoteQueuedTaskIfWorkflowStepHasCapacity(ctx, task, fromStepID, destinationStepID, limit)
	return true, claimed, err
}

func (s *Service) promoteFeederQueuedTask(ctx context.Context, candidate *models.Task, fromStepID, oldWorkflowID string, targetStep *wfmodels.WorkflowStep, position int, skipped map[string]struct{}, sessionID string) bool {
	oldState := candidate.State
	if candidate.Metadata == nil {
		candidate.Metadata = make(map[string]interface{})
	}
	candidate.WIPAdmitted = true
	candidate.QueuedForStepID = ""
	candidate.QueuedAt = nil
	candidate.Metadata[models.MetaKeyQueuePromotionPending] = map[string]interface{}{
		"from_step_id": fromStepID,
	}
	candidate.Position = position
	candidate.WorkflowID = targetStep.WorkflowID
	candidate.WorkflowStepID = targetStep.ID
	if err := s.syncTaskStateForQueuePromotion(ctx, candidate, targetStep); err != nil {
		s.logger.Warn("failed to prepare feeder queued promotion", zap.String("task_id", candidate.ID), zap.Error(err))
		skipped[candidate.ID] = struct{}{}
		return s.promoteNextQueuedTask(ctx, targetStep, position, skipped)
	}
	if promoter, ok := s.tasks.(workflowQueuedTaskPromoter); ok {
		claimed, err := promoter.PromoteQueuedTaskIfWorkflowStepHasCapacity(ctx, candidate, fromStepID, targetStep.ID, targetStep.WIPLimit)
		if err != nil {
			s.logger.Warn("failed to promote feeder queued task", zap.String("task_id", candidate.ID), zap.Error(err))
			return false
		}
		if !claimed {
			skipped[candidate.ID] = struct{}{}
			return s.promoteNextQueuedTask(ctx, targetStep, position, skipped)
		}
		s.publishTaskEvent(ctx, events.TaskUpdated, candidate, nil, oldWorkflowID)
		if oldState != candidate.State {
			s.publishTaskEvent(ctx, events.TaskStateChanged, candidate, &oldState)
		}
		s.recordQueuedPromotion(ctx, candidate.ID, fromStepID, targetStep.ID)
		s.publishTaskMovedEvent(ctx, candidate, oldWorkflowID, fromStepID, targetStep.ID, sessionID, "")
		return true
	} else if admissionRepo, ok := s.tasks.(workflowMoveAdmissionRepository); ok {
		claimed, err := admissionRepo.UpdateTaskWithWorkflowStepAdmission(ctx, candidate, fromStepID, targetStep.ID, targetStep.WIPLimit)
		if err != nil {
			s.logger.Warn("failed to promote feeder queued task", zap.String("task_id", candidate.ID), zap.Error(err))
			skipped[candidate.ID] = struct{}{}
			return s.promoteNextQueuedTask(ctx, targetStep, position, skipped)
		}
		if !claimed {
			skipped[candidate.ID] = struct{}{}
			return s.promoteNextQueuedTask(ctx, targetStep, position, skipped)
		}
		s.publishTaskEvent(ctx, events.TaskUpdated, candidate, nil, oldWorkflowID)
		if oldState != candidate.State {
			s.publishTaskEvent(ctx, events.TaskStateChanged, candidate, &oldState)
		}
		s.recordQueuedPromotion(ctx, candidate.ID, fromStepID, targetStep.ID)
		s.publishTaskMovedEvent(ctx, candidate, oldWorkflowID, fromStepID, targetStep.ID, sessionID, "")
		return true
	}
	// ctx here still carries the identity of whoever triggered the move that
	// freed the slot, so MoveTaskWithOptions' authorizeTaskID applies to the
	// promoted candidate too. That is safe by construction: a step's
	// PullFromStepID resolves within the same workflow, a workflow belongs to
	// one workspace, and a workspace has one owner — so the candidate always
	// belongs to the caller who just passed the same check.
	//
	// If a future configuration ever allowed cross-workspace feeder pulls, this
	// would refuse and log below rather than promote. Leave it that way: do not
	// strip the identity to "fix" it. Promoting another user's task into a step
	// they cannot see is the worse outcome, and the atomic promoter above (which
	// the SQLite repository implements, so it is the only path production takes)
	// does not go through a guarded method at all.
	if _, err := s.MoveTaskWithOptions(ctx, candidate.ID, targetStep.WorkflowID, targetStep.ID, position, MoveTaskOptions{PreserveDeferredLaunch: true}); err != nil {
		skipped[candidate.ID] = struct{}{}
		s.logger.Warn("skipping queued task that could not be promoted", zap.String("task_id", candidate.ID), zap.String("to_step_id", targetStep.ID), zap.Error(err))
		return s.promoteNextQueuedTask(ctx, targetStep, position, skipped)
	}
	return true
}

func (s *Service) recordQueuedPromotion(ctx context.Context, taskID, fromStepID, toStepID string) {
	if s.stepHistoryRecorder == nil {
		return
	}
	session := s.resolvePrimaryOrActiveSession(ctx, taskID)
	if session == nil {
		return
	}
	if asyncRecorder, ok := s.stepHistoryRecorder.(asyncStepHistoryRecorder); ok {
		if asyncRecorder.EnqueueStepTransition(session.ID, fromStepID, toStepID, wfmodels.StepTransitionTriggerQueuePromotion, nil, nil) {
			return
		}
	}
	writeCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), constants.StepHistoryWriteTimeout)
	defer cancel()
	if err := s.stepHistoryRecorder.CreateStepTransition(writeCtx, session.ID, fromStepID, toStepID, wfmodels.StepTransitionTriggerQueuePromotion, nil, nil); err != nil {
		s.logger.Warn("failed to record queued task promotion", zap.String("task_id", taskID), zap.Error(err))
	}
}

func (s *Service) feederCandidateSession(ctx context.Context, taskID string) (*models.TaskSession, bool) {
	sessions, err := s.sessions.ListTaskSessions(ctx, taskID)
	if err != nil {
		s.logger.Warn("skipping feeder task after active session lookup failed", zap.String("task_id", taskID), zap.Error(err))
		return nil, true
	}
	var primary, fallback *models.TaskSession
	for _, session := range sessions {
		if session == nil {
			continue
		}
		if isSessionMoveBlocked(session.State) {
			return nil, true
		}
		if !isSessionActive(session.State) {
			continue
		}
		if session.IsPrimary && primary == nil {
			primary = session
		}
		if !session.IsPrimary && fallback == nil {
			fallback = session
		}
	}
	if primary != nil {
		return primary, false
	}
	return fallback, false
}

func (s *Service) nextQueuedCandidate(ctx context.Context, targetStep *wfmodels.WorkflowStep, skipped map[string]struct{}) (*models.Task, error) {
	candidates, err := s.tasks.ListTasksByWorkflowStep(ctx, targetStep.ID)
	if err != nil {
		return nil, err
	}
	sameStep := findSameStepQueuedCandidate(candidates, targetStep.ID, skipped)
	feederTask, err := s.nextFeederQueuedCandidate(ctx, targetStep, skipped)
	if err != nil {
		return nil, err
	}
	if sameStep == nil {
		return feederTask, nil
	}
	if sameStep != nil {
		return sameStep, nil
	}
	return feederTask, nil
}

func findSameStepQueuedCandidate(candidates []*models.Task, stepID string, skipped map[string]struct{}) *models.Task {
	var selected *models.Task
	for _, candidate := range candidates {
		if candidate == nil || candidate.WIPAdmitted || candidate.QueuedForStepID != stepID {
			continue
		}
		if _, seen := skipped[candidate.ID]; seen {
			continue
		}
		if selected == nil || queuedTaskBefore(candidate, selected) {
			selected = candidate
		}
	}
	return selected
}

func (s *Service) nextFeederQueuedCandidate(
	ctx context.Context,
	targetStep *wfmodels.WorkflowStep,
	skipped map[string]struct{},
) (*models.Task, error) {
	if targetStep.PullFromStepID == "" {
		return nil, nil
	}
	excluded := skippedTaskIDs(skipped)
	if pullRepo, ok := s.tasks.(workflowQueuedPullRepository); ok {
		candidate, err := pullRepo.NextQueuedTaskForStepExcluding(
			ctx, targetStep.PullFromStepID, targetStep.ID, excluded,
		)
		if err != nil || candidate != nil {
			return candidate, err
		}
	}
	if legacyPullRepo, ok := s.tasks.(workflowPullRepository); ok {
		return legacyPullRepo.NextPullCandidateExcluding(ctx, targetStep.PullFromStepID, excluded)
	}
	return nil, nil
}

// queuedTaskBefore is the WIP promotion comparator
// (REQ-TASKS-KANBAN-TASK-REORDERING-001.1, .36): position, priority rank,
// queued_at (coalesced to created_at when absent), created_at, id. Delegates
// to models.StepOrderLess, the single source of truth this comparator's
// byte-identical orchestrator copy and the reorder repository also use.
func queuedTaskBefore(left, right *models.Task) bool {
	return models.StepOrderLess(left, right)
}

func skippedTaskIDs(skipped map[string]struct{}) []string {
	ids := make([]string, 0, len(skipped))
	for id := range skipped {
		if id != "" {
			ids = append(ids, id)
		}
	}
	return ids
}

func (s *Service) updateMovedTask(
	ctx context.Context,
	task *models.Task,
	oldStepID string,
	targetStep *wfmodels.WorkflowStep,
	admittedState *v1.TaskState,
	opts MoveTaskOptions,
) (bool, error) {
	if targetStep == nil || oldStepID == targetStep.ID {
		return s.updateMovedTaskSameStep(ctx, task, opts)
	}
	return s.updateMovedTaskCrossStep(ctx, task, oldStepID, targetStep, admittedState, opts)
}

// updateMovedTaskSameStep handles the no-step-change branch of updateMovedTask
// (a reorder, or a plugin move that names the task's current step): it writes
// through the plain repository UpdateTask, not one of the WIP-admission call
// sites, since there is no admission decision to make when the step is
// unchanged.
func (s *Service) updateMovedTaskSameStep(ctx context.Context, task *models.Task, opts MoveTaskOptions) (bool, error) {
	if opts.ExpectedWorkflowID != nil {
		// Same-step writes go through plain UpdateTask, which has no
		// expected-workflow parameter (it is the general-purpose writer
		// used by many unrelated call sites). A caller carrying a CAS
		// guard must route through the narrower method instead, or the
		// guard would silently stop applying for the same-step case —
		// see workflowMoveConflictRepository's doc.
		conflictRepo, ok := s.tasks.(workflowMoveConflictRepository)
		if !ok {
			return false, fmt.Errorf("workflow conflict guard repository unavailable for task %s", task.ID)
		}
		if err := conflictRepo.UpdateTaskIfWorkflowMatches(ctx, task, *opts.ExpectedWorkflowID); err != nil {
			return false, err
		}
		return task.WIPAdmitted, nil
	}
	if err := s.tasks.UpdateTaskPreservingDeferredLaunch(ctx, task); err != nil {
		return false, err
	}
	return task.WIPAdmitted, nil
}

// updateMovedTaskCrossStep handles the step-changing branch of
// updateMovedTask: it runs the target step's WIP admission decision (queuing
// the task instead of moving it when the step is at capacity) and persists
// the result.
func (s *Service) updateMovedTaskCrossStep(
	ctx context.Context,
	task *models.Task,
	oldStepID string,
	targetStep *wfmodels.WorkflowStep,
	admittedState *v1.TaskState,
	opts MoveTaskOptions,
) (bool, error) {
	admissionRepo, ok := s.tasks.(workflowMoveAdmissionRepository)
	if !ok {
		return false, fmt.Errorf("workflow step admission repository unavailable for step %s", targetStep.ID)
	}
	expectedWorkflowID := ""
	if opts.ExpectedWorkflowID != nil {
		expectedWorkflowID = *opts.ExpectedWorkflowID
	}
	if admissionWithState, ok := s.tasks.(workflowMoveAdmissionWithStateRepository); ok {
		return admissionWithState.UpdateTaskWithWorkflowStepAdmissionAndState(
			ctx, task, oldStepID, targetStep.ID, targetStep.WIPLimit, admittedState, true, expectedWorkflowID,
		)
	}
	if expectedWorkflowID != "" {
		// The CAS-guarded caller (plugin moves) always runs against a
		// production repository, which implements the atomic variant above.
		// A repository that only exposes the legacy admission method has no
		// way to honor the guard, so fail closed instead of silently
		// dropping it.
		return false, fmt.Errorf("workflow conflict guard unavailable on legacy admission repository for step %s", targetStep.ID)
	}

	// Keep compatibility with narrow test/dry-run repositories that expose
	// only the original admission method. Production repositories implement the
	// atomic variant above, so this fallback is never used for real moves.
	admitted, err := admissionRepo.UpdateTaskWithWorkflowStepAdmission(ctx, task, oldStepID, targetStep.ID, targetStep.WIPLimit)
	if err != nil {
		return false, err
	}
	if admitted && admittedState != nil {
		task.State = *admittedState
		delete(task.Metadata, models.MetaKeyQueuedMoveExitPending)
	} else if !admitted {
		if task.Metadata == nil {
			task.Metadata = make(map[string]interface{})
		}
		task.Metadata[models.MetaKeyQueuedMoveExitPending] = true
	}
	if err := s.tasks.UpdateTask(ctx, task); err != nil {
		return false, err
	}
	return admitted, nil
}

func (s *Service) syncTaskStateForQueuePromotion(ctx context.Context, task *models.Task, targetStep *wfmodels.WorkflowStep) error {
	if targetStep == nil {
		return nil
	}
	terminal, err := s.terminalWorkflowStep(ctx, targetStep.ID)
	if err != nil {
		return fmt.Errorf("sync promoted task state for %s: %w", task.ID, err)
	}
	if terminal {
		if !models.IsTerminalTaskState(task.State) {
			task.State = v1.TaskStateCompleted
		}
		return nil
	}
	if task.State == v1.TaskStateCompleted {
		task.State = v1.TaskStateTODO
	}
	return nil
}

func (s *Service) validateTaskMove(ctx context.Context, task *models.Task, workflowID, workflowStepID string, opts MoveTaskOptions) (*wfmodels.WorkflowStep, error) {
	if task.ArchivedAt != nil {
		return nil, fmt.Errorf("archived tasks cannot be moved")
	}
	if err := s.validateMoveSessions(ctx, task.ID, opts); err != nil {
		return nil, err
	}
	targetWorkflow, err := s.workflows.GetWorkflow(ctx, workflowID)
	if err != nil {
		return nil, fmt.Errorf("failed to get target workflow: %w", err)
	}
	if targetWorkflow.WorkspaceID != task.WorkspaceID {
		return nil, fmt.Errorf("target workflow is in a different workspace")
	}
	if s.workflowStepGetter == nil {
		return nil, nil
	}
	targetStep, err := s.workflowStepGetter.GetStep(ctx, workflowStepID)
	if err != nil {
		return nil, fmt.Errorf("failed to get target workflow step: %w", err)
	}
	if targetStep.WorkflowID != workflowID {
		return nil, fmt.Errorf("target workflow step does not belong to target workflow")
	}
	return targetStep, nil
}

func (s *Service) validateMoveSessions(ctx context.Context, taskID string, opts MoveTaskOptions) error {
	sessions, err := s.sessions.ListTaskSessions(ctx, taskID)
	if err != nil {
		return fmt.Errorf("failed to list task sessions: %w", err)
	}
	for _, session := range sessions {
		if isSessionMoveBlocked(session.State) {
			if opts.AllowActivePrimarySession && session.IsPrimary {
				continue
			}
			return fmt.Errorf("task has an active session (%s)", session.State)
		}
	}
	return nil
}

func isSessionMoveBlocked(state models.TaskSessionState) bool {
	return state == models.TaskSessionStateStarting ||
		state == models.TaskSessionStateRunning
}

// resolvePrimaryOrActiveSession returns the primary session if it is in an active
// state, otherwise falls back to the most recently started active session.
func (s *Service) resolvePrimaryOrActiveSession(ctx context.Context, taskID string) *models.TaskSession {
	primary, _ := s.sessions.GetPrimarySessionByTaskID(ctx, taskID)
	if primary != nil && isSessionActive(primary.State) {
		return primary
	}
	active, err := s.sessions.GetActiveTaskSessionByTaskID(ctx, taskID)
	if err != nil || active == nil {
		return nil
	}
	return active
}

// recordManualStepTransition writes the ADR 0015 audit row for a
// user/agent-initiated move. trigger is normally StepTransitionTriggerManual;
// callers driving an approval-gated transition (ApproveSession) pass
// StepTransitionTriggerApproval — a zero value defaults to Manual. It is a
// no-op when no recorder is wired or when the task has no session to record
// against — session_step_history.session_id is a NOT NULL FK to
// task_sessions, so a session-less move cannot be recorded without a
// schema change. Runtime writes use the workflow service's bounded worker;
// failures are logged and swallowed because this is best-effort telemetry.
func (s *Service) recordManualStepTransition(ctx context.Context, sessionID, fromStepID, toStepID string, trigger wfmodels.StepTransitionTrigger, actors ...wfmodels.StepTransitionActor) {
	if s.stepHistoryRecorder == nil {
		return
	}
	if sessionID == "" {
		s.logger.Debug("skipping manual step transition audit: task has no session",
			zap.String("from_step_id", fromStepID),
			zap.String("to_step_id", toStepID))
		return
	}
	if trigger == "" {
		trigger = wfmodels.StepTransitionTriggerManual
	}
	var actorID *string
	actor := wfmodels.StepTransitionActorHuman
	if len(actors) > 0 {
		actor = actors[0]
	}
	if actor == wfmodels.StepTransitionActorHuman {
		if identity, ok := authn.IdentityFromContext(ctx); ok && identity.UserID != "" {
			actorID = &identity.UserID
		}
	}
	if asyncRecorder, ok := s.stepHistoryRecorder.(asyncStepHistoryRecorder); ok {
		if asyncRecorder.EnqueueStepTransition(sessionID, fromStepID, toStepID, trigger, actorID, nil) {
			return
		}
	}
	// The step change is already durably persisted by the time this runs.
	// Use a detached, bounded context so a cancelled request context (client
	// disconnect, turn-end) cannot drop the audit row for a transition that
	// already committed.
	writeCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), constants.StepHistoryWriteTimeout)
	defer cancel()
	if err := s.stepHistoryRecorder.CreateStepTransition(
		writeCtx, sessionID, fromStepID, toStepID, trigger, actorID, nil,
	); err != nil {
		s.logger.Warn("failed to record manual step transition",
			zap.String("session_id", sessionID),
			zap.String("from_step_id", fromStepID),
			zap.String("to_step_id", toStepID),
			zap.Error(err))
	}
}

func isSessionActive(state models.TaskSessionState) bool {
	return state == models.TaskSessionStateCreated ||
		state == models.TaskSessionStateStarting ||
		state == models.TaskSessionStateRunning ||
		state == models.TaskSessionStateWaitingForInput
}

// CountTasksByWorkflow returns the number of tasks in a workflow.
func (s *Service) CountTasksByWorkflow(ctx context.Context, workflowID string) (int, error) {
	if err := s.authorizeWorkflowID(ctx, workflowID); err != nil {
		return 0, err
	}
	return s.tasks.CountTasksByWorkflow(ctx, workflowID)
}

// CountTasksByWorkflowStep returns the number of tasks in a workflow step.
func (s *Service) CountTasksByWorkflowStep(ctx context.Context, stepID string) (int, error) {
	if _, scoped := callerScope(ctx); scoped {
		if s.workflowStepGetter == nil {
			return 0, errors.New("workflow step getter unavailable")
		}
		step, err := s.workflowStepGetter.GetStep(ctx, stepID)
		if err != nil {
			return 0, err
		}
		if step == nil {
			return 0, repoerrors.ErrWorkspaceNotFound
		}
		if err := s.authorizeWorkflowID(ctx, step.WorkflowID); err != nil {
			return 0, err
		}
	}
	return s.tasks.CountTasksByWorkflowStep(ctx, stepID)
}

// BulkMoveTasksResult contains the result of a BulkMoveTasks operation.
type BulkMoveTasksResult struct {
	MovedCount int
}

// BulkMoveSelectedTasks moves an explicit task list to a target workflow step.
// The list order is treated as the visible UI order; tasks already in the
// target step are skipped. Validation reads tasks one at a time because the UI
// sends small selected batches; the move is not transactional if task state
// changes between pre-validation and an individual MoveTask call.
func (s *Service) BulkMoveSelectedTasks(ctx context.Context, taskIDs []string, targetWorkflowID, targetStepID string) (*BulkMoveTasksResult, error) {
	ids := uniqueTaskIDs(taskIDs)
	if len(ids) == 0 {
		return &BulkMoveTasksResult{MovedCount: 0}, nil
	}
	bulkActorKind, bulkActorID := steptelemetry.HumanOrSystemActor(ctx)
	ctx = steptelemetry.WithAttribution(ctx, steptelemetry.Attribution{
		Trigger: steptelemetry.TriggerBulkMove, ActorKind: bulkActorKind, ActorID: bulkActorID,
	})

	tasks, err := s.validateSelectedMoveBatch(ctx, ids, targetWorkflowID, targetStepID)
	if err != nil {
		return nil, err
	}
	// The server now computes each arriving task's position from the target
	// step's current max (REQ-TASKS-KANBAN-TASK-REORDERING-001.28), so the
	// literal passed to MoveTask below is ignored and no longer needs
	// precomputing. What still matters is dispatch order: MoveTask is called
	// once per task, sequentially, so the final order is simply the call
	// order (REQ-TASKS-KANBAN-TASK-REORDERING-001.29) — source step ordinal
	// ascending, then within one source step that step's admitted band in
	// step order followed by its queued band in step order.
	orderedTasks, err := s.orderTasksForBulkMove(ctx, tasks)
	if err != nil {
		return nil, err
	}

	// Hold every step this batch will touch — the target plus each task's
	// own source step, known now that orderedTasks is resolved — as one
	// ascending-ordered lock set across the whole dispatch loop below, not
	// just within each individual MoveTask call's own transaction. A
	// per-call-only lock leaves a window between two of this batch's own
	// calls where an unrelated arrival (another create, move, WIP promotion,
	// or automatic transition) into targetStepID can land in the middle of
	// the batch's sequence, breaking the batch-scoped consecutiveness
	// REQ-TASKS-KANBAN-TASK-REORDERING-001.29 requires. Locking only the
	// target here and letting each per-task MoveTask acquire its own source
	// step deadlocks against an ordinary single move running the opposite
	// direction between the same two steps.
	lockedCtx, lockedTasks, unlock, err := s.acquireBulkMoveStepLocks(ctx, orderedTasks, targetStepID)
	if err != nil {
		return nil, err
	}
	ctx = lockedCtx
	defer unlock()
	// The lock acquisition above may have re-read a task's step after a
	// concurrent drift, so re-derive dispatch order from that corrected
	// membership rather than the pre-lock orderedTasks — otherwise a task
	// whose stale record still shows it at targetStepID is silently skipped
	// below even though it has since left, and any task whose real source
	// step changed keeps the wrong REQ-TASKS-KANBAN-TASK-REORDERING-001.29
	// submission position.
	orderedTasks, err = s.orderTasksForBulkMove(ctx, lockedTasks)
	if err != nil {
		return nil, err
	}
	if s.bulkMoveAfterLockForTest != nil {
		s.bulkMoveAfterLockForTest()
	}

	movedCount := 0
	for _, task := range orderedTasks {
		if task.WorkflowID == targetWorkflowID && task.WorkflowStepID == targetStepID {
			continue
		}
		if _, err := s.MoveTask(ctx, task.ID, targetWorkflowID, targetStepID, 0); err != nil {
			return nil, fmt.Errorf("failed to move task %s: %w", task.ID, err)
		}
		movedCount++
		if s.bulkMoveAfterTaskForTest != nil {
			s.bulkMoveAfterTaskForTest()
		}
	}

	return &BulkMoveTasksResult{MovedCount: movedCount}, nil
}

// stepArrivalBatchLocker is the narrow capability BulkMoveSelectedTasks and
// BulkMoveTasks need from the task repository to hold a batch's whole step
// set — target plus every distinct source step — locked across their
// sequential dispatch loop, following the same runtime-asserted
// narrow-interface pattern as reorderRepository.
type stepArrivalBatchLocker interface {
	LockStepArrivalsForBatch(ctx context.Context, stepIDs ...string) (context.Context, func())
}

// bulkMoveLockStepIDs returns the target step plus every distinct source
// step among tasks, for a stepArrivalBatchLocker call. Duplicates and the
// empty string are harmless: withStepArrivalLocks dedupes and sorts before
// acquiring.
func bulkMoveLockStepIDs(tasks []*models.Task, targetStepID string) []string {
	ids := make([]string, 0, len(tasks)+1)
	ids = append(ids, targetStepID)
	for _, task := range tasks {
		ids = append(ids, task.WorkflowStepID)
	}
	return ids
}

// bulkMoveLockRetryLimit bounds acquireBulkMoveStepLocks' re-read/retry loop.
// A source step can only drift a bounded number of times before some other
// caller's own work stalls behind this batch's locks, so a low limit is
// sufficient to converge on the real, uncontended case while still failing
// loudly instead of spinning forever if something is pathologically racing
// this batch on every attempt.
const bulkMoveLockRetryLimit = 8

// acquireBulkMoveStepLocks locks every step BulkMoveSelectedTasks/
// BulkMoveTasks will touch — the target step plus each task's current
// source step — using bulkMoveLockStepIDs' set for a stepArrivalBatchLocker.
//
// tasks is read once, before any lock is held, to compute that initial
// step set. A source step named by that pre-lock read can still change
// between the read and lock acquisition (another mover wins the race),
// which would leave this batch holding a stale source step's lock while a
// concurrent move on the true current source step tries to lock the target
// step this batch already holds — the AB-BA deadlock this exists to avoid.
// So once the lock is held, tasks are re-read by ID under it; if any
// task's step moved, the stale lock set is released and a fresh set is
// acquired for the corrected steps, repeating until the read matches the
// locked set or bulkMoveLockRetryLimit is exhausted.
//
// The returned task slice is the membership the lock was finally acquired
// against, not the pre-lock tasks argument. Callers must dispatch and derive
// REQ-TASKS-KANBAN-TASK-REORDERING-001.29 submission order from this
// returned slice, never from their own pre-lock read: a task that drifted
// out of the target step during acquisition is only reflected here, and a
// caller still using its pre-lock copy would treat that task as if it had
// never left.
func (s *Service) acquireBulkMoveStepLocks(
	ctx context.Context, tasks []*models.Task, targetStepID string,
) (context.Context, []*models.Task, func(), error) {
	locker, ok := s.tasks.(stepArrivalBatchLocker)
	if !ok {
		return ctx, tasks, func() {}, nil
	}
	if s.bulkMoveBeforeLockForTest != nil {
		s.bulkMoveBeforeLockForTest()
	}
	taskIDs := make([]string, len(tasks))
	for i, task := range tasks {
		taskIDs[i] = task.ID
	}
	current := tasks
	for attempt := 0; attempt < bulkMoveLockRetryLimit; attempt++ {
		lockedCtx, unlock := locker.LockStepArrivalsForBatch(ctx, bulkMoveLockStepIDs(current, targetStepID)...)
		fresh, err := s.tasks.GetTasksByIDs(lockedCtx, taskIDs)
		if err != nil {
			unlock()
			return ctx, nil, func() {}, fmt.Errorf("failed to verify bulk move lock set: %w", err)
		}
		if bulkMoveTaskStepsMatch(current, fresh) {
			return lockedCtx, fresh, unlock, nil
		}
		unlock()
		current = fresh
	}
	return ctx, nil, func() {}, fmt.Errorf("bulk move step lock set did not stabilize after %d attempts", bulkMoveLockRetryLimit)
}

// bulkMoveTaskStepsMatch reports whether a and b agree on every task's
// current WorkflowStepID, keyed by task ID rather than slice position since
// GetTasksByIDs does not guarantee the requested order.
func bulkMoveTaskStepsMatch(a, b []*models.Task) bool {
	if len(a) != len(b) {
		return false
	}
	stepByID := make(map[string]string, len(a))
	for _, task := range a {
		stepByID[task.ID] = task.WorkflowStepID
	}
	for _, task := range b {
		if stepByID[task.ID] != task.WorkflowStepID {
			return false
		}
	}
	return true
}

// orderTasksForBulkMove re-derives the
// REQ-TASKS-KANBAN-TASK-REORDERING-001.29 submission order from each task's
// current source step, rather than trusting the caller-supplied list order:
// source step ordinal ascending, ties on source step id ascending
// (a selection can span workflows, so two source steps can share an
// ordinal), then within one source step that step's admitted band in step
// order followed by its queued band in step order.
func (s *Service) orderTasksForBulkMove(ctx context.Context, tasks []*models.Task) ([]*models.Task, error) {
	stepOrdinals := make(map[string]int, len(tasks))
	if s.workflowStepGetter == nil {
		// No ordinal source: fall back to submission order rather than
		// aborting the batch — every task groups under its own step with an
		// equal (zero) ordinal, so bulkMoveSubmissionOrder's per-step
		// StepOrderLess sort still applies within each source step.
		return bulkMoveSubmissionOrder(tasks, stepOrdinals), nil
	}
	for _, task := range tasks {
		if _, ok := stepOrdinals[task.WorkflowStepID]; ok {
			continue
		}
		step, err := s.workflowStepGetter.GetStep(ctx, task.WorkflowStepID)
		if err != nil {
			// An empty or dangling source step (e.g. deleted concurrently)
			// must not abort every other task's move: order it last rather
			// than failing the whole batch.
			stepOrdinals[task.WorkflowStepID] = math.MaxInt32
			continue
		}
		stepOrdinals[task.WorkflowStepID] = step.Position
	}
	return bulkMoveSubmissionOrder(tasks, stepOrdinals), nil
}

func bulkMoveSubmissionOrder(tasks []*models.Task, stepOrdinals map[string]int) []*models.Task {
	type sourceGroup struct {
		stepID  string
		ordinal int
		tasks   []*models.Task
	}
	groupsByStep := make(map[string]*sourceGroup, len(tasks))
	var groups []*sourceGroup
	for _, task := range tasks {
		group, ok := groupsByStep[task.WorkflowStepID]
		if !ok {
			group = &sourceGroup{stepID: task.WorkflowStepID, ordinal: stepOrdinals[task.WorkflowStepID]}
			groupsByStep[task.WorkflowStepID] = group
			groups = append(groups, group)
		}
		group.tasks = append(group.tasks, task)
	}
	sort.SliceStable(groups, func(i, j int) bool {
		if groups[i].ordinal != groups[j].ordinal {
			return groups[i].ordinal < groups[j].ordinal
		}
		return groups[i].stepID < groups[j].stepID
	})

	ordered := make([]*models.Task, 0, len(tasks))
	for _, group := range groups {
		admitted, queued := bulkMoveBandSplit(group.tasks, group.stepID)
		sort.SliceStable(admitted, func(i, j int) bool { return models.StepOrderLess(admitted[i], admitted[j]) })
		sort.SliceStable(queued, func(i, j int) bool { return models.StepOrderLess(queued[i], queued[j]) })
		ordered = append(ordered, admitted...)
		ordered = append(ordered, queued...)
	}
	return ordered
}

// bulkMoveBandSplit mirrors the reorder repository's band partition
// (Terminology: the queued band is !wip_admitted && queued_for_step_id ==
// stepID; everything else in the step is the admitted band).
func bulkMoveBandSplit(tasks []*models.Task, stepID string) (admitted, queued []*models.Task) {
	for _, task := range tasks {
		if !task.WIPAdmitted && task.QueuedForStepID == stepID {
			queued = append(queued, task)
		} else {
			admitted = append(admitted, task)
		}
	}
	return admitted, queued
}

func uniqueTaskIDs(taskIDs []string) []string {
	seen := make(map[string]struct{}, len(taskIDs))
	result := make([]string, 0, len(taskIDs))
	for _, id := range taskIDs {
		if id == "" {
			continue
		}
		if _, ok := seen[id]; ok {
			continue
		}
		seen[id] = struct{}{}
		result = append(result, id)
	}
	return result
}

func (s *Service) validateSelectedMoveBatch(ctx context.Context, taskIDs []string, targetWorkflowID, targetStepID string) ([]*models.Task, error) {
	tasks := make([]*models.Task, 0, len(taskIDs))
	for _, id := range taskIDs {
		task, err := s.tasks.GetTask(ctx, id)
		if err != nil {
			return nil, err
		}
		if task.WorkflowID != targetWorkflowID || task.WorkflowStepID != targetStepID {
			if _, err := s.validateTaskMove(ctx, task, targetWorkflowID, targetStepID, MoveTaskOptions{}); err != nil {
				return nil, fmt.Errorf("task %s cannot be moved: %w", id, err)
			}
		}
		tasks = append(tasks, task)
	}
	return tasks, nil
}

// BulkMoveTasks moves all tasks from a source workflow/step to a target workflow/step.
// If sourceStepID is empty, all tasks in the source workflow are moved.
func (s *Service) BulkMoveTasks(ctx context.Context, sourceWorkflowID, sourceStepID, targetWorkflowID, targetStepID string) (*BulkMoveTasksResult, error) {
	bulkActorKind, bulkActorID := steptelemetry.HumanOrSystemActor(ctx)
	ctx = steptelemetry.WithAttribution(ctx, steptelemetry.Attribution{
		Trigger: steptelemetry.TriggerBulkMove, ActorKind: bulkActorKind, ActorID: bulkActorID,
	})

	// Get the tasks to move
	var tasks []*models.Task
	var err error
	if sourceStepID != "" {
		tasks, err = s.tasks.ListTasksByWorkflowStep(ctx, sourceStepID)
	} else {
		tasks, err = s.tasks.ListTasks(ctx, sourceWorkflowID)
	}
	if err != nil {
		return nil, fmt.Errorf("failed to list tasks for bulk move: %w", err)
	}

	if len(tasks) == 0 {
		return &BulkMoveTasksResult{MovedCount: 0}, nil
	}

	// Re-derive the REQ-TASKS-KANBAN-TASK-REORDERING-001.29 submission order
	// from each task's source step, the same as BulkMoveSelectedTasks: the
	// server now computes each arriving task's position from the target
	// step's current max, so dispatch order alone decides the final order.
	orderedTasks, err := s.orderTasksForBulkMove(ctx, tasks)
	if err != nil {
		return nil, err
	}

	// Hold every step this batch will touch, for the same reason and in the
	// same ascending-ordered way as BulkMoveSelectedTasks — see its lock
	// call for the deadlock this avoids.
	lockedCtx, lockedTasks, unlock, err := s.acquireBulkMoveStepLocks(ctx, orderedTasks, targetStepID)
	if err != nil {
		return nil, err
	}
	ctx = lockedCtx
	defer unlock()
	// See BulkMoveSelectedTasks's identical re-derivation: the lock
	// acquisition above may have corrected a task's step after a concurrent
	// drift, so dispatch order must be re-derived from that membership
	// rather than the pre-lock orderedTasks.
	orderedTasks, err = s.orderTasksForBulkMove(ctx, lockedTasks)
	if err != nil {
		return nil, err
	}
	if s.bulkMoveAfterLockForTest != nil {
		s.bulkMoveAfterLockForTest()
	}

	for _, task := range orderedTasks {
		if _, err := s.MoveTask(ctx, task.ID, targetWorkflowID, targetStepID, 0); err != nil {
			return nil, fmt.Errorf("failed to move task %s: %w", task.ID, err)
		}
		if s.bulkMoveAfterTaskForTest != nil {
			s.bulkMoveAfterTaskForTest()
		}
	}

	s.logger.Info("bulk moved tasks",
		zap.String("source_workflow_id", sourceWorkflowID),
		zap.String("source_step_id", sourceStepID),
		zap.String("target_workflow_id", targetWorkflowID),
		zap.String("target_step_id", targetStepID),
		zap.Int("moved_count", len(tasks)))

	return &BulkMoveTasksResult{MovedCount: len(tasks)}, nil
}