src/allocation.py

"""Task allocation engine for dynamic workload management."""

import hashlib
import json
from datetime import datetime, timezone
from typing import Any

from .database import DB


class AllocationValidator:
    """Input validator for allocation operations."""

    @staticmethod
    def parse_amount_spec(spec: str) -> dict[str, Any]:
        """Parse amount specification.

        Examples:
            'count:100' -> {'type': 'count', 'value': 100}
            'ratio:0.5' -> {'type': 'ratio', 'value': 0.5}
            'equal' -> {'type': 'equal'}
            'ratio:A:0.3,B:0.7' -> {'type': 'ratio_per_user', 'users': {'A': 0.3, 'B': 0.7}}
            'all' -> {'type': 'all'}

        """
        if spec == 'equal':
            return {'type': 'equal'}

        if spec == 'all':
            return {'type': 'all'}

        if spec.startswith('count:'):
            count = int(spec[6:])
            if count <= 0:
                raise ValueError(f'Count must be positive, got: {count}')
            return {'type': 'count', 'value': count}

        if spec.startswith('ratio:'):
            rest = spec[6:]
            # Check if per-user ratio
            if ':' in rest:
                # Format: ratio:A:0.3,B:0.3,C:0.4
                user_ratios = {}
                for part in rest.split(','):
                    user, ratio_str = part.split(':')
                    ratio = float(ratio_str)
                    if ratio < 0 or ratio > 1:
                        raise ValueError(f'Ratio must be in [0, 1], got: {ratio} for user {user}')
                    user_ratios[user.strip()] = ratio

                total = sum(user_ratios.values())
                if abs(total - 1.0) > 0.001:
                    raise ValueError(f'User ratios must sum to 1.0, got: {total}')

                return {'type': 'ratio_per_user', 'users': user_ratios}
            else:
                # Single ratio
                ratio = float(rest)
                if ratio <= 0 or ratio > 1:
                    raise ValueError(f'Ratio must be in (0, 1], got: {ratio}')
                return {'type': 'ratio', 'value': ratio}

        raise ValueError(f'Invalid amount spec: {spec}')

    @staticmethod
    def parse_target(target: str) -> dict[str, Any]:
        """Parse target specification.

        Examples:
            'user:A' -> {'type': 'single', 'user': 'A'}
            'users:A,B,C' -> {'type': 'multiple', 'users': ['A', 'B', 'C']}

        """
        if target.startswith('user:'):
            user = target[5:].strip()
            if not user:
                raise ValueError('User ID cannot be empty')
            return {'type': 'single', 'user': user}

        if target.startswith('users:'):
            users = [u.strip() for u in target[6:].split(',')]
            users = list(dict.fromkeys(users))  # Remove duplicates, preserve order
            if not users:
                raise ValueError('User list cannot be empty')
            if len(users) != len(target[6:].split(',')):
                print('Warning: Duplicate users removed from target')
            return {'type': 'multiple', 'users': users}

        raise ValueError(f'Invalid target: {target}')

    @staticmethod
    def validate_no_self_overlap(from_users: list[str], to_users: list[str]) -> list[str]:
        """Remove self-overlapping users and return filtered list."""
        overlap = set(from_users) & set(to_users)
        if overlap:
            print(f'Warning: Removing self-overlapping users: {overlap}')
            return [u for u in to_users if u not in overlap]
        return to_users

    @staticmethod
    def check_users_exist(db: DB, round_id: str, users: list[str]) -> None:
        """Check if users exist in this round (warn for new users)."""
        with db._connect() as conn:
            # Check if assignments table exists
            cur = conn.execute("SELECT to_regclass('public.assignments')")
            if cur.fetchone()[0] is None:
                return  # New schema not initialized, skip check

            for user in users:
                cur = conn.execute(
                    'SELECT COUNT(*) FROM assignments WHERE round_id = %s AND user_id = %s', (round_id, user)
                )
                count = cur.fetchone()[0]
                if count == 0:
                    print(f"Info: User '{user}' is new to this round")


class AllocationEngine:
    """Task allocation engine for managing workload distribution.

    Methods:
    - allocate(from_source, to_target, amount_spec, redundancy, force, dry_run, reason)
    - release(from_user, amount_spec, dry_run, reason)
    - rebalance(users, mode_spec, dry_run, reason)

    """

    def __init__(
        self,
        db: DB,
        round_id: str,
        operator: str = 'admin',
    ):
        """Initialize allocation engine.

        Args:
            db: Database instance
            round_id: Round ID
            operator: Operator name for audit logging

        """
        self.db = db
        self.round_id = round_id
        self.operator = operator

    def allocate(
        self,
        from_source: str,
        to_target: str,
        amount_spec: str,
        redundancy: int,
        force: bool = False,
        dry_run: bool = False,
        reason: str = '',
    ) -> dict[str, Any]:
        """Core allocation function.

        Redundancy design:
        - Each task needs N different annotators (redundancy parameter)
        - Each user can only annotate a task ONCE (enforced by PRIMARY KEY)
        - If a user already has a task, it cannot be assigned to them again
        - This may reduce the actual number of tasks allocated if users already
          have some of the source tasks

        Args:
            from_source: Source of tasks ('unassigned', 'user:A')
            to_target: Target users ('user:A', 'users:A,B,C')
            amount_spec: Amount specification ('count:100', 'ratio:0.5', 'equal')
            redundancy: Number of different people who should label each task
            force: Allow transferring in_progress tasks (default: False)
            dry_run: Preview only, don't execute (default: False)
            reason: Reason for this allocation (for audit)

        Returns:
            {
                'success': bool,
                'affected_users': list[str],
                'affected_count': int,
                'distribution': dict[str, int],  # user -> count
                'item_ids_sample': list[str],
                'error': str  # only if success=False
            }

        """
        # 1. Validate inputs
        validator = AllocationValidator()
        amount_parsed = validator.parse_amount_spec(amount_spec)
        target_parsed = validator.parse_target(to_target)

        # 2. Get target users
        target_users = self._get_target_users(target_parsed)
        if not target_users:
            return {'success': False, 'error': 'No target users specified', 'affected_count': 0}

        validator.check_users_exist(self.db, self.round_id, target_users)

        # 3. Get available source tasks for EACH user (filtered by redundancy at SQL level)
        user_available_tasks = self._get_available_source_items(from_source, target_users, redundancy, force)

        # Check if any user has available tasks
        total_available = sum(len(tasks) for tasks in user_available_tasks.values())
        if total_available == 0:
            return {
                'success': False,
                'error': f'No available tasks from source {from_source} for target users '
                '(tasks may already be assigned or at redundancy limit)',
                'affected_count': 0,
            }

        # 4. Calculate distribution considering each user's available tasks
        distribution = self._calculate_distribution_per_user(user_available_tasks, amount_parsed)

        # 6. Execute or preview
        if dry_run:
            result = self._preview_allocation(distribution)
        else:
            result = self._execute_allocation(from_source, to_target, amount_spec, distribution, force, reason)

        return result

    def release(self, from_user: str, amount_spec: str, dry_run: bool = False, reason: str = '') -> dict[str, Any]:
        """Release tasks from a user back to unassigned pool.

        Args:
            from_user: User ID to release tasks from
            amount_spec: Amount to release ('count:50', 'ratio:0.3', 'all')
            dry_run: Preview only
            reason: Reason for release

        Returns:
            Result dictionary with success status

        """
        validator = AllocationValidator()
        amount_parsed = validator.parse_amount_spec(amount_spec)

        # Get user's pending tasks
        with self.db._connect() as conn:
            cur = conn.execute(
                """
                SELECT item_id
                FROM assignments
                WHERE round_id = %s AND user_id = %s AND status = 'pending'
                ORDER BY assigned_at
                """,
                (self.round_id, from_user),
            )
            pending_items = [row[0] for row in cur.fetchall()]

        if not pending_items:
            return {'success': False, 'error': 'No pending tasks to release'}

        # Calculate how many to release
        if amount_parsed['type'] == 'count':
            to_release = pending_items[: min(amount_parsed['value'], len(pending_items))]
        elif amount_parsed['type'] == 'ratio':
            count = int(len(pending_items) * amount_parsed['value'])
            to_release = pending_items[:count]
        elif amount_parsed['type'] == 'all':
            to_release = pending_items
        else:
            return {'success': False, 'error': f'Invalid amount_spec for release: {amount_spec}'}

        # Preview release
        if dry_run:
            return {
                'success': True,
                'dry_run': True,
                'affected_users': [from_user],
                'affected_count': len(to_release),
                'item_ids_sample': to_release[:10],
            }

        # Execute release
        now = datetime.now(timezone.utc).isoformat()
        with self.db._connect() as conn, self.db._lock:
            placeholders = ','.join(['%s'] * len(to_release))
            conn.execute(
                f"""
                DELETE FROM assignments
                WHERE round_id = %s AND user_id = %s AND item_id IN ({placeholders})
                """,
                (self.round_id, from_user, *to_release),
            )

            # Record log
            conn.execute(
                """
                INSERT INTO allocation_history
                (round_id, operation, operator, from_source, to_target, amount_spec,
                 affected_users, affected_count, item_ids_sample, item_ids_hash,
                 dry_run, force, reason, created_at)
                VALUES (%s, 'release', %s, %s, 'pool', %s, %s, %s, %s, %s, 0, 0, %s, %s)
                """,
                (
                    self.round_id,
                    self.operator,
                    f'user:{from_user}',
                    amount_spec,
                    json.dumps([from_user]),
                    len(to_release),
                    json.dumps(to_release[:10]),
                    self._compute_item_hash(to_release),
                    reason,
                    now,
                ),
            )

        return {
            'success': True,
            'affected_users': [from_user],
            'affected_count': len(to_release),
            'item_ids_sample': to_release[:10],
        }

    def rebalance(self, users: list[str], mode_spec: str, dry_run: bool = False, reason: str = '') -> dict[str, Any]:
        """Rebalance pending tasks among selected users.

        Args:
            users: List of user IDs to rebalance among
            mode_spec: Rebalance mode ('equal', 'ratio:A:0.3,B:0.7')
            dry_run: Preview only
            reason: Reason for rebalance

        Returns:
            Result dictionary with success status

        """
        # 1. Collect all unique pending tasks from these users (avoid duplicates for redundancy > 1)
        with self.db._connect() as conn:
            placeholders_users = ','.join(['%s'] * len(users))
            cur = conn.execute(
                f"""
                SELECT DISTINCT item_id
                FROM assignments
                WHERE round_id = %s AND user_id IN ({placeholders_users}) AND status = 'pending'
                """,
                (self.round_id, *users),
            )
            all_pending = [row[0] for row in cur.fetchall()]

        if not all_pending:
            return {'success': False, 'error': 'No pending tasks to rebalance'}

        # 2. Check for redundancy > 1 tasks and collect their current assignment counts
        task_redundancy_info = {}
        with self.db._connect() as conn:
            for item_id in all_pending:
                # Get redundancy config
                cur = conn.execute(
                    """
                    SELECT redundancy_required
                    FROM task_config
                    WHERE round_id = %s AND item_id = %s
                    """,
                    (self.round_id, item_id),
                )
                row = cur.fetchone()
                redundancy_required = row[0] if row else 1

                # Count current assignments among rebalance users
                placeholders = ','.join(['%s'] * len(users))
                cur = conn.execute(
                    f"""
                    SELECT COUNT(*)
                    FROM assignments
                    WHERE round_id = %s AND item_id = %s AND user_id IN ({placeholders}) AND status = 'pending'
                    """,
                    (self.round_id, item_id, *users),
                )
                current_count = cur.fetchone()[0]

                task_redundancy_info[item_id] = {
                    'redundancy_required': redundancy_required,
                    'current_count': current_count,
                }

        # 3. Parse distribution mode
        validator = AllocationValidator()
        amount_parsed = validator.parse_amount_spec(mode_spec)

        # 4. Warn if there are redundancy > 1 tasks
        redundancy_gt1_tasks = [
            item_id for item_id, info in task_redundancy_info.items() if info['redundancy_required'] > 1
        ]
        if redundancy_gt1_tasks:
            print(
                f'Warning: {len(redundancy_gt1_tasks)} tasks have redundancy > 1. '
                'Rebalance will redistribute unique tasks (may change redundancy counts). '
                f'Consider using allocate/release for fine control. Sample: {redundancy_gt1_tasks[:3]}'
            )

        # 5. Build per-user available pools from all_pending (exclude items the user already has)
        #    We do NOT filter by redundancy limit here because we are redistributing
        #    the same set of tasks within the same user group; distinct-user counts
        #    will remain within limits. We only prevent giving a user a task they
        #    already have (any status).
        user_available_tasks: dict[str, list[str]] = {}
        with self.db._connect() as conn:
            for user in users:
                if not all_pending:
                    user_available_tasks[user] = []
                    continue
                # exclude items that the user already has
                placeholders = ','.join(['%s'] * len(all_pending))
                cur = conn.execute(
                    f"""
                    SELECT x.item_id
                    FROM (
                        SELECT %s AS round_id, %s AS user_id
                    ) p
                    JOIN (
                        SELECT item_id FROM (
                            VALUES {','.join(['(%s)'] * len(all_pending))}
                        ) AS v(item_id)
                    ) AS x ON 1=1
                    WHERE NOT EXISTS (
                        SELECT 1
                        FROM assignments a
                        WHERE a.round_id = p.round_id
                          AND a.item_id = x.item_id
                          AND a.user_id = p.user_id
                    )
                    """,
                    (self.round_id, user, *all_pending),
                )
                user_available_tasks[user] = [row[0] for row in cur.fetchall()]

        # 6. Calculate new distribution using per-user pools
        distribution = self._calculate_distribution_per_user(user_available_tasks, amount_parsed)

        # Preview rebalance
        if dry_run:
            return {
                'success': True,
                'dry_run': True,
                'affected_users': users,
                'affected_count': len(all_pending),
                'distribution': {u: len(items) for u, items in distribution.items()},
                'item_ids_sample': all_pending[:10],
            }

        # 7. Execute rebalance (delete old assignments, insert new ones)
        now = datetime.now(timezone.utc).isoformat()
        with self.db._connect() as conn, self.db._lock:
            conn.execute('BEGIN')

            try:
                # Delete all pending tasks from these users
                placeholders_users = ','.join(['%s'] * len(users))
                conn.execute(
                    f"""
                    DELETE FROM assignments
                    WHERE round_id = %s AND user_id IN ({placeholders_users}) AND status = 'pending'
                    """,
                    (self.round_id, *users),
                )

                # Insert new assignments
                for user, item_ids in distribution.items():
                    for item_id in item_ids:
                        # Check if this user already has this task (avoid PRIMARY KEY violation)
                        # Each user can only have ONE assignment per task
                        cur = conn.execute(
                            """
                            SELECT 1
                            FROM assignments
                            WHERE round_id = %s AND item_id = %s AND user_id = %s
                            """,
                            (self.round_id, item_id, user),
                        )
                        if cur.fetchone():
                            # User already has this task, skip
                            continue

                        # Compute next slot based on how many DISTINCT users already have this task
                        cur = conn.execute(
                            """
                            SELECT COUNT(DISTINCT user_id)
                            FROM assignments
                            WHERE round_id = %s AND item_id = %s
                            """,
                            (self.round_id, item_id),
                        )
                        next_slot = cur.fetchone()[0]

                        # Insert new assignment
                        conn.execute(
                            """
                            INSERT INTO assignments
                            (round_id, item_id, user_id, status, assigned_at, redundancy_slot)
                            VALUES (%s, %s, %s, 'pending', %s, %s)
                            """,
                            (self.round_id, item_id, user, now, next_slot),
                        )

                # Record log
                conn.execute(
                    """
                    INSERT INTO allocation_history
                    (round_id, operation, operator, from_source, to_target, amount_spec,
                     affected_users, affected_count, item_ids_sample, item_ids_hash,
                     dry_run, force, reason, created_at)
                    VALUES (%s, 'rebalance', %s, %s, %s, %s, %s, %s, %s, %s, 0, 0, %s, %s)
                    """,
                    (
                        self.round_id,
                        self.operator,
                        f'users:{",".join(users)}',
                        f'users:{",".join(users)}',
                        mode_spec,
                        json.dumps(users),
                        len(all_pending),
                        json.dumps(all_pending[:10]),
                        self._compute_item_hash(all_pending),
                        reason,
                        now,
                    ),
                )

                conn.execute('COMMIT')

                return {
                    'success': True,
                    'affected_users': users,
                    'affected_count': len(all_pending),
                    'distribution': {u: len(items) for u, items in distribution.items()},
                }

            except Exception as e:
                conn.execute('ROLLBACK')
                raise e

    # ========== Internal Methods ==========

    def _get_available_source_items(
        self, from_source: str, target_users: list[str], redundancy: int, force: bool
    ) -> dict[str, list[str]]:
        """Get available source tasks for EACH target user separately.

        This method returns a dictionary mapping each user to their available tasks.
        Different users have different available tasks because they
        already have different existing assignments.

        For each user, filters at SQL level to return tasks that:
        1. Come from the specified source (unassigned pool or another user)
        2. Are not already assigned to THIS SPECIFIC USER
        3. Have not reached their redundancy limit

        Args:
            from_source: Source specification ('unassigned' or 'user:UserID')
            target_users: List of users who will receive these tasks
            redundancy: Number of different annotators required per task
            force: If True, allow transferring in_progress tasks (only for user source)

        Returns:
            Dictionary mapping user_id to list of available item_ids
            Example: {'Alice': ['task1', 'task2'], 'Bob': ['task1', 'task3'], ...}

        """
        user_available_tasks = {}

        with self.db._connect() as conn:
            # Query available tasks for EACH user separately
            for user in target_users:
                if from_source == 'unassigned':
                    # Get tasks from unassigned pool that this user doesn't have
                    cur = conn.execute(
                        """
                        SELECT t.item_id
                        FROM tasks t
                        LEFT JOIN task_config tc
                          ON t.round_id = tc.round_id AND t.item_id = tc.item_id
                        WHERE t.round_id = %s
                          -- This specific user doesn't have this task
                          AND NOT EXISTS (
                              SELECT 1
                              FROM assignments a
                              WHERE a.round_id = t.round_id
                                AND a.item_id = t.item_id
                                AND a.user_id = %s
                          )
                          -- Check redundancy limit (current assignments < required)
                          AND (
                              SELECT COUNT(DISTINCT a2.user_id)
                              FROM assignments a2
                              WHERE a2.round_id = t.round_id AND a2.item_id = t.item_id
                          ) < COALESCE(tc.redundancy_required, %s)
                          -- Check redundancy completion
                          AND COALESCE(tc.redundancy_completed, 0) < COALESCE(tc.redundancy_required, %s)
                        ORDER BY t.order_key
                        """,
                        (self.round_id, user, redundancy, redundancy),
                    )
                elif from_source.startswith('user:'):
                    source_user_id = from_source[5:]

                    if force:
                        status_filter = "('pending', 'in_progress')"
                    else:
                        status_filter = "('pending')"

                    cur = conn.execute(
                        f"""
                        SELECT DISTINCT a1.item_id
                        FROM assignments a1
                        LEFT JOIN task_config tc
                          ON a1.round_id = tc.round_id AND tc.item_id = a1.item_id
                        WHERE a1.round_id = %s
                          AND a1.user_id = %s
                          AND a1.status IN {status_filter}
                          -- This specific target user doesn't have this task
                          AND NOT EXISTS (
                              SELECT 1
                              FROM assignments a2
                              WHERE a2.round_id = a1.round_id
                                AND a2.item_id = a1.item_id
                                AND a2.user_id = %s
                          )
                          -- Check redundancy limit
                          AND (
                              SELECT COUNT(DISTINCT a3.user_id)
                              FROM assignments a3
                              WHERE a3.round_id = a1.round_id AND a3.item_id = a1.item_id
                          ) < COALESCE(tc.redundancy_required, %s)
                          -- Check redundancy completion
                          AND COALESCE(tc.redundancy_completed, 0) < COALESCE(tc.redundancy_required, %s)
                        ORDER BY a1.assigned_at
                        """,
                        (self.round_id, source_user_id, user, redundancy, redundancy),
                    )
                else:
                    raise ValueError(f'Invalid from_source: {from_source}')

                user_available_tasks[user] = [row[0] for row in cur.fetchall()]

        return user_available_tasks

    def _get_target_users(self, target_parsed: dict) -> list[str]:
        """Get target user list."""
        if target_parsed['type'] == 'single':
            return [target_parsed['user']]
        else:
            return target_parsed['users']

    def _calculate_distribution_per_user(
        self, user_available_tasks: dict[str, list[str]], amount_parsed: dict
    ) -> dict[str, list[str]]:
        """Calculate task distribution for each user based on their available tasks.

        Key insight: Each user has a different pool of available tasks because they
        have different existing assignments. This method respects that and assigns
        tasks from each user's individual available pool.

        Args:
            user_available_tasks: Dict mapping user_id to list of available task IDs
                                  Example: {'Alice': ['task1', 'task2', 'task3'],
                                           'Bob': ['task1', 'task4']}
            amount_parsed: Parsed amount specification

        Returns:
            Dict mapping user_id to list of assigned task IDs
            Example: {'Alice': ['task1', 'task2'], 'Bob': ['task1']}

        """
        target_users = list(user_available_tasks.keys())
        distribution = {}

        if amount_parsed['type'] == 'count':
            # Each user gets up to N tasks from their available pool
            count = amount_parsed['value']
            for user in target_users:
                available = user_available_tasks[user]
                distribution[user] = available[: min(count, len(available))]

        elif amount_parsed['type'] == 'ratio':
            # Each user gets X% of their available tasks
            ratio = amount_parsed['value']
            for user in target_users:
                available = user_available_tasks[user]
                count = int(len(available) * ratio)
                distribution[user] = available[:count]

        elif amount_parsed['type'] == 'equal':
            # Try to give each user roughly equal number of tasks
            # Strategy: Calculate target count as average, but respect each user's limit
            total_available = sum(len(tasks) for tasks in user_available_tasks.values())
            target_per_user = total_available // len(target_users)

            for user in target_users:
                available = user_available_tasks[user]
                distribution[user] = available[: min(target_per_user, len(available))]

        elif amount_parsed['type'] == 'ratio_per_user':
            # User-specific ratios: calculate from total available pool
            user_ratios = amount_parsed['users']
            total_available = sum(len(tasks) for tasks in user_available_tasks.values())

            # Sort by user_id for deterministic behavior
            sorted_users = sorted(user_ratios.items(), key=lambda x: x[0])

            for user, ratio in sorted_users:
                if user not in user_available_tasks:
                    distribution[user] = []
                    continue

                available = user_available_tasks[user]
                target_count = int(total_available * ratio)
                # Take up to target_count, but limited by what's available for this user
                distribution[user] = available[: min(target_count, len(available))]

        elif amount_parsed['type'] == 'all':
            # Give each user all their available tasks
            for user in target_users:
                distribution[user] = user_available_tasks[user]

        else:
            raise ValueError(f'Unknown amount type: {amount_parsed["type"]}')

        return distribution

    def _calculate_distribution(
        self, item_ids: list[str], target_users: list[str], amount_parsed: dict
    ) -> dict[str, list[str]]:
        """Calculate task distribution for each user (old method, kept for compatibility).

        Returns:
            {'user_A': ['item_1', 'item_2'], 'user_B': ['item_3'], ...}

        """
        n_items = len(item_ids)

        if amount_parsed['type'] == 'count':
            count = min(amount_parsed['value'], n_items)
            selected_items = item_ids[:count]
            return self._distribute_equal(selected_items, target_users)

        elif amount_parsed['type'] == 'ratio':
            ratio = amount_parsed['value']
            count = int(n_items * ratio)
            selected_items = item_ids[:count]
            return self._distribute_equal(selected_items, target_users)

        elif amount_parsed['type'] in ('equal', 'all'):
            return self._distribute_equal(item_ids, target_users)

        elif amount_parsed['type'] == 'ratio_per_user':
            user_ratios = amount_parsed['users']
            distribution = {}
            start_idx = 0

            # Sort by user_id for deterministic behavior
            sorted_users = sorted(user_ratios.items(), key=lambda x: x[0])

            for i, (user, ratio) in enumerate(sorted_users):
                if i == len(sorted_users) - 1:
                    # Last user gets all remaining (avoid floating point errors)
                    distribution[user] = item_ids[start_idx:]
                else:
                    count = int(n_items * ratio)
                    distribution[user] = item_ids[start_idx : start_idx + count]
                    start_idx += count

            return distribution

        raise ValueError(f'Unknown amount type: {amount_parsed["type"]}')

    def _distribute_equal(self, item_ids: list[str], users: list[str]) -> dict[str, list[str]]:
        """Distribute tasks equally among users (deterministic remainder allocation)."""
        n_items = len(item_ids)
        n_users = len(users)
        base_count = n_items // n_users
        remainder = n_items % n_users

        # Sort by user_id for deterministic behavior
        sorted_users = sorted(users)

        distribution = {}
        start_idx = 0

        for i, user in enumerate(sorted_users):
            # First 'remainder' users get one extra item
            count = base_count + (1 if i < remainder else 0)
            distribution[user] = item_ids[start_idx : start_idx + count]
            start_idx += count

        return distribution

    def _compute_item_hash(self, item_ids: list[str]) -> str:
        """Compute hash of item_id list for audit."""
        content = ','.join(sorted(item_ids))
        return hashlib.sha256(content.encode()).hexdigest()[:16]

    def _preview_allocation(self, distribution: dict[str, list[str]]) -> dict[str, Any]:
        """Dry-run preview."""
        total_allocated = sum(len(items) for items in distribution.values())

        # Sample: first 2 items from each user
        sample_ids = []
        for items in distribution.values():
            sample_ids.extend(items[:2])
        sample_ids = sample_ids[:10]

        return {
            'success': True,
            'dry_run': True,
            'affected_users': list(distribution.keys()),
            'affected_count': total_allocated,
            'distribution': {u: len(items) for u, items in distribution.items()},
            'item_ids_sample': sample_ids,
            'item_ids_hash': self._compute_item_hash([iid for items in distribution.values() for iid in items]),
        }

    def _execute_allocation(
        self,
        from_source: str,
        to_target: str,
        amount_spec: str,
        distribution: dict[str, list[str]],
        force: bool,
        reason: str,
    ) -> dict[str, Any]:
        """Execute actual allocation."""
        now = datetime.now(timezone.utc).isoformat()
        all_assigned_items = [iid for items in distribution.values() for iid in items]

        with self.db._connect() as conn, self.db._lock:
            conn.execute('BEGIN')

            try:
                # 1. If transferring from a user, delete their assignments (no skipped state)
                if from_source.startswith('user:'):
                    source_user = from_source[5:]
                    placeholders = ','.join(['%s'] * len(all_assigned_items))
                    conn.execute(
                        f"""
                        DELETE FROM assignments
                        WHERE round_id = %s AND user_id = %s AND item_id IN ({placeholders})
                        """,
                        (self.round_id, source_user, *all_assigned_items),
                    )

                # 2. Create assignments for target users
                for user, item_ids in distribution.items():
                    for item_id in item_ids:
                        # Check if this user already has this task (avoid PRIMARY KEY violation)
                        # Each user can only have ONE assignment per task
                        cur = conn.execute(
                            """
                            SELECT 1
                            FROM assignments
                            WHERE round_id = %s AND item_id = %s AND user_id = %s
                            """,
                            (self.round_id, item_id, user),
                        )
                        if cur.fetchone():
                            # User already has this task, skip
                            continue

                        # Compute next slot based on how many DISTINCT users already have this task
                        cur = conn.execute(
                            """
                            SELECT COUNT(DISTINCT user_id)
                            FROM assignments
                            WHERE round_id = %s AND item_id = %s
                            """,
                            (self.round_id, item_id),
                        )
                        next_slot = cur.fetchone()[0]

                        conn.execute(
                            """
                            INSERT INTO assignments
                            (round_id, item_id, user_id, status, assigned_at, redundancy_slot)
                            VALUES (%s, %s, %s, 'pending', %s, %s)
                            """,
                            (self.round_id, item_id, user, now, next_slot),
                        )

                # 3. Record log
                conn.execute(
                    """
                    INSERT INTO allocation_history
                    (round_id, operation, operator, from_source, to_target, amount_spec,
                     affected_users, affected_count, item_ids_sample, item_ids_hash,
                     dry_run, force, reason, created_at)
                    VALUES (%s, 'allocate', %s, %s, %s, %s, %s, %s, %s, %s, 0, %s, %s, %s)
                    """,
                    (
                        self.round_id,
                        self.operator,
                        from_source,
                        to_target,
                        amount_spec,
                        json.dumps(list(distribution.keys())),
                        len(all_assigned_items),
                        json.dumps(all_assigned_items[:10]),
                        self._compute_item_hash(all_assigned_items),
                        1 if force else 0,
                        reason,
                        now,
                    ),
                )

                conn.execute('COMMIT')

                return {
                    'success': True,
                    'affected_users': list(distribution.keys()),
                    'affected_count': len(all_assigned_items),
                    'distribution': {u: len(items) for u, items in distribution.items()},
                    'item_ids_sample': all_assigned_items[:10],
                }

            except Exception as e:
                conn.execute('ROLLBACK')
                raise e