FieldMapper

field_mapper

Field mapping between beads and ClickUp.

CustomFieldMapper

Maps custom fields between beads issues and ClickUp tasks.

Handles type-aware conversion for different ClickUp custom field types: - drop_down: Maps string values to dropdown option IDs - text/short_text: Direct string mapping - url: URL string mapping - number: Numeric value mapping - date: Unix timestamp mapping - users: User ID mapping - labels: Multi-select label mapping

Source code in beads_clickup/field_mapper.py

class CustomFieldMapper:
    """Maps custom fields between beads issues and ClickUp tasks.

    Handles type-aware conversion for different ClickUp custom field types:
    - drop_down: Maps string values to dropdown option IDs
    - text/short_text: Direct string mapping
    - url: URL string mapping
    - number: Numeric value mapping
    - date: Unix timestamp mapping
    - users: User ID mapping
    - labels: Multi-select label mapping
    """

    def __init__(self, config: Dict[str, Any]):
        """Initialize custom field mapper with configuration.

        Args:
            config: Configuration dictionary from config.yaml
        """
        self.config = config
        self.field_mapping = config.get("field_mapping", {})
        self.custom_fields = self.field_mapping.get("custom_fields", {})
        self.lenient_mode = self.field_mapping.get("lenient_mode", True)

        # Cache for dropdown options (populated during field discovery)
        self._dropdown_options_cache: Dict[str, Dict[str, str]] = {}

    def set_dropdown_options(self, field_name: str, options: Dict[str, str]):
        """Cache dropdown options for a field.

        Args:
            field_name: Beads field name (e.g., 'lifecycle_stage')
            options: Dict mapping option names to option IDs
        """
        self._dropdown_options_cache[field_name] = options

    def get_field_config(self, field_name: str) -> Optional[Dict[str, Any]]:
        """Get configuration for a custom field.

        Handles both exact matches and normalized matches (strips numeric prefixes).

        Args:
            field_name: Beads field name

        Returns:
            Field configuration dict or None if not configured
        """
        import re

        # Try exact match first
        if field_name in self.custom_fields:
            return self.custom_fields[field_name]

        # Try normalized name (strip numeric prefixes)
        normalized = re.sub(r'^\d+_', '', field_name)
        for config_name, config in self.custom_fields.items():
            config_normalized = re.sub(r'^\d+_', '', config_name)
            if config_normalized == normalized:
                return config

        return None

    def get_clickup_field_id(self, field_name: str) -> Optional[str]:
        """Get ClickUp field ID for a beads field name.

        Handles both exact matches and normalized matches (strips numeric prefixes).

        Args:
            field_name: Beads field name

        Returns:
            ClickUp custom field UUID or None
        """
        import re

        # Try exact match first
        field_config = self.get_field_config(field_name)
        if field_config:
            return field_config.get("clickup_field_id")

        # Try normalized name (strip numeric prefixes)
        normalized = re.sub(r'^\d+_', '', field_name)
        for config_name, config in self.custom_fields.items():
            config_normalized = re.sub(r'^\d+_', '', config_name)
            if config_normalized == normalized:
                return config.get("clickup_field_id")

        return None

    def convert_to_clickup_value(
        self,
        field_name: str,
        beads_value: Any,
    ) -> Optional[Any]:
        """Convert a beads field value to ClickUp custom field format.

        Args:
            field_name: Beads field name
            beads_value: Value from beads issue

        Returns:
            Value formatted for ClickUp API, or None if conversion fails
        """
        if beads_value is None:
            return None

        field_config = self.get_field_config(field_name)
        if not field_config:
            if self.lenient_mode:
                logger.debug("Unknown custom field '%s', skipping", field_name)
                return None
            raise ValueError(f"Unknown custom field: {field_name}")

        field_type = field_config.get("type", "text")

        try:
            if field_type == "drop_down":
                return self._convert_dropdown_to_clickup(field_name, beads_value)
            elif field_type in ("text", "short_text"):
                return str(beads_value)
            elif field_type == "url":
                return str(beads_value)
            elif field_type == "number":
                return float(beads_value)
            elif field_type == "date":
                # Expect Unix timestamp in milliseconds
                return int(beads_value)
            elif field_type == "users":
                # Expect user ID or list of user IDs
                if isinstance(beads_value, list):
                    return beads_value
                return [beads_value]
            elif field_type == "labels":
                # Multi-select labels
                if isinstance(beads_value, list):
                    return beads_value
                return [beads_value]
            else:
                logger.warning(
                    "Unknown field type '%s' for field '%s'", field_type, field_name
                )
                return str(beads_value)

        except Exception as e:
            if self.lenient_mode:
                logger.warning(
                    "Failed to convert field '%s' value '%s': %s",
                    field_name,
                    beads_value,
                    e,
                )
                return None
            raise

    def _convert_dropdown_to_clickup(
        self,
        field_name: str,
        beads_value: str,
    ) -> Optional[Union[str, int]]:
        """Convert a beads value to ClickUp dropdown option.

        Args:
            field_name: Beads field name
            beads_value: String value (option name)

        Returns:
            ClickUp option ID or orderindex, or None if not found
        """
        # Check cache for option ID mapping (populated by API discovery)
        if field_name in self._dropdown_options_cache:
            options = self._dropdown_options_cache[field_name]
            # Try exact match first
            if beads_value in options:
                return options[beads_value]
            # Try case-insensitive match
            beads_lower = beads_value.lower()
            for opt_name, opt_id in options.items():
                if opt_name.lower() == beads_lower:
                    return opt_id

            logger.warning(
                "Dropdown option '%s' not found in API cache for field '%s' "
                "(available: %s). Trying config fallback.",
                beads_value,
                field_name,
                list(options.keys()),
            )
            # Fall through to config-based fallback below

        # Fallback: use config options list to determine orderindex
        field_config = self.get_field_config(field_name)
        if field_config:
            config_options = field_config.get("options", [])
            # Try exact match
            for idx, opt in enumerate(config_options):
                if opt == beads_value:
                    logger.info(
                        "Resolved '%s' for field '%s' via config orderindex %d",
                        beads_value, field_name, idx,
                    )
                    return idx
            # Try case-insensitive match
            beads_lower = beads_value.lower() if isinstance(beads_value, str) else ""
            for idx, opt in enumerate(config_options):
                if isinstance(opt, str) and opt.lower() == beads_lower:
                    logger.info(
                        "Resolved '%s' for field '%s' via config orderindex %d (case-insensitive)",
                        beads_value, field_name, idx,
                    )
                    return idx

        if self.lenient_mode:
            logger.warning(
                "Dropdown option '%s' not found for field '%s' in cache or config",
                beads_value,
                field_name,
            )
            return None
        raise ValueError(
            f"Invalid dropdown option '{beads_value}' for field '{field_name}'"
        )

    def convert_from_clickup_value(
        self,
        field_name: str,
        clickup_value: Any,
        field_type: Optional[str] = None,
    ) -> Optional[Any]:
        """Convert a ClickUp custom field value to beads format.

        Args:
            field_name: Beads field name
            clickup_value: Value from ClickUp task
            field_type: Override field type (useful when type is known from API)

        Returns:
            Value formatted for beads, or None if conversion fails
        """
        if clickup_value is None:
            return None

        field_config = self.get_field_config(field_name)
        if not field_config and not field_type:
            if self.lenient_mode:
                logger.debug("Unknown custom field '%s', skipping", field_name)
                return None
            raise ValueError(f"Unknown custom field: {field_name}")

        ftype = field_type or (field_config.get("type") if field_config else "text")

        try:
            if ftype == "drop_down":
                return self._convert_dropdown_from_clickup(field_name, clickup_value)
            elif ftype in ("text", "short_text"):
                return str(clickup_value) if clickup_value else None
            elif ftype == "url":
                return str(clickup_value) if clickup_value else None
            elif ftype == "number":
                return float(clickup_value) if clickup_value else None
            elif ftype == "date":
                return int(clickup_value) if clickup_value else None
            elif ftype == "users":
                # ClickUp returns users as list of dicts with 'id' key
                if isinstance(clickup_value, list):
                    return [
                        u.get("id") if isinstance(u, dict) else u for u in clickup_value
                    ]
                return clickup_value
            elif ftype == "labels":
                if isinstance(clickup_value, list):
                    return clickup_value
                return [clickup_value] if clickup_value else []
            else:
                return clickup_value

        except Exception as e:
            if self.lenient_mode:
                logger.warning(
                    "Failed to convert ClickUp field '%s' value '%s': %s",
                    field_name,
                    clickup_value,
                    e,
                )
                return None
            raise

    def _convert_dropdown_from_clickup(
        self,
        field_name: str,
        clickup_value: Any,
    ) -> Optional[str]:
        """Convert a ClickUp dropdown value to beads string.

        Args:
            field_name: Beads field name
            clickup_value: ClickUp dropdown value (could be int index or dict)

        Returns:
            Option name string or None
        """
        # ClickUp returns dropdown as dict with 'name' or as orderindex/ID
        if isinstance(clickup_value, dict):
            return clickup_value.get("name") or clickup_value.get("value")
        elif isinstance(clickup_value, (int, str)):
            # It's an ID (int orderindex or string UUID) - reverse lookup in cache
            if field_name in self._dropdown_options_cache:
                options = self._dropdown_options_cache[field_name]
                for opt_name, opt_id in options.items():
                    if opt_id == clickup_value or str(opt_id) == str(clickup_value):
                        return opt_name
        return str(clickup_value) if clickup_value else None

    def beads_custom_fields_to_clickup(
        self,
        beads_fields: Dict[str, Any],
    ) -> List[Dict[str, Any]]:
        """Convert beads custom fields dict to ClickUp custom_fields array.

        Args:
            beads_fields: Dict of field_name -> value

        Returns:
            List of dicts with 'id' and 'value' for ClickUp API
        """
        clickup_fields = []

        for field_name, beads_value in beads_fields.items():
            field_id = self.get_clickup_field_id(field_name)
            if not field_id:
                if self.lenient_mode:
                    logger.debug("No ClickUp field ID for '%s', skipping", field_name)
                    continue
                raise ValueError(f"No ClickUp field ID configured for: {field_name}")

            clickup_value = self.convert_to_clickup_value(field_name, beads_value)
            if clickup_value is not None:
                clickup_fields.append(
                    {
                        "id": field_id,
                        "value": clickup_value,
                    }
                )

        return clickup_fields

    def clickup_custom_fields_to_beads(
        self,
        clickup_fields: List[Dict[str, Any]],
    ) -> Dict[str, Any]:
        """Convert ClickUp custom_fields array to beads fields dict.

        Args:
            clickup_fields: List of ClickUp custom field dicts

        Returns:
            Dict of field_name -> value for beads
        """
        beads_fields = {}

        # Build reverse lookup: clickup_field_id -> beads_field_name
        id_to_name = {}
        for field_name, field_config in self.custom_fields.items():
            field_id = field_config.get("clickup_field_id")
            if field_id:
                id_to_name[field_id] = field_name

        for cu_field in clickup_fields:
            field_id = cu_field.get("id")
            field_name = id_to_name.get(field_id)

            if not field_name:
                # Try to use the ClickUp field name as fallback
                cu_name = cu_field.get("name", "").lower().replace(" ", "_")
                if cu_name in self.custom_fields:
                    field_name = cu_name
                else:
                    logger.debug("Unknown ClickUp field ID '%s', skipping", field_id)
                    continue

            cu_value = cu_field.get("value")
            field_type = cu_field.get("type")
            beads_value = self.convert_from_clickup_value(
                field_name, cu_value, field_type
            )

            if beads_value is not None:
                beads_fields[field_name] = beads_value

        return beads_fields

    def get_required_fields_with_defaults(self) -> Dict[str, Any]:
        """Get all required fields and their default values.

        Returns:
            Dict of field_name -> default_value for required fields
        """
        defaults = {}
        for field_name, field_config in self.custom_fields.items():
            if field_config.get("required", False):
                default = field_config.get("default")
                if default is not None:
                    defaults[field_name] = default
        return defaults

    def validate_required_fields(
        self,
        beads_fields: Dict[str, Any],
    ) -> Dict[str, Any]:
        """Validate and fill in required fields with defaults.

        Args:
            beads_fields: Dict of field_name -> value

        Returns:
            Updated dict with defaults filled in for missing required fields

        Raises:
            ValueError: If required field missing and no default (strict mode only)
        """
        result = dict(beads_fields)

        for field_name, field_config in self.custom_fields.items():
            if not field_config.get("required", False):
                continue

            if field_name not in result or result[field_name] is None:
                default = field_config.get("default")
                if default is not None:
                    result[field_name] = default
                    logger.debug(
                        "Using default value '%s' for required field '%s'",
                        default,
                        field_name,
                    )
                elif not self.lenient_mode:
                    raise ValueError(
                        f"Required field '{field_name}' missing and no default configured"
                    )

        return result

__init__(config)

Initialize custom field mapper with configuration.

Parameters:

Name	Type	Description	Default
config	Dict[str, Any]	Configuration dictionary from config.yaml	required

Source code in beads_clickup/field_mapper.py

def __init__(self, config: Dict[str, Any]):
    """Initialize custom field mapper with configuration.

    Args:
        config: Configuration dictionary from config.yaml
    """
    self.config = config
    self.field_mapping = config.get("field_mapping", {})
    self.custom_fields = self.field_mapping.get("custom_fields", {})
    self.lenient_mode = self.field_mapping.get("lenient_mode", True)

    # Cache for dropdown options (populated during field discovery)
    self._dropdown_options_cache: Dict[str, Dict[str, str]] = {}

beads_custom_fields_to_clickup(beads_fields)

Convert beads custom fields dict to ClickUp custom_fields array.

Parameters:

Name	Type	Description	Default
beads_fields	Dict[str, Any]	Dict of field_name -> value	required

Returns:

Type	Description
List[Dict[str, Any]]	List of dicts with 'id' and 'value' for ClickUp API

Source code in beads_clickup/field_mapper.py

def beads_custom_fields_to_clickup(
    self,
    beads_fields: Dict[str, Any],
) -> List[Dict[str, Any]]:
    """Convert beads custom fields dict to ClickUp custom_fields array.

    Args:
        beads_fields: Dict of field_name -> value

    Returns:
        List of dicts with 'id' and 'value' for ClickUp API
    """
    clickup_fields = []

    for field_name, beads_value in beads_fields.items():
        field_id = self.get_clickup_field_id(field_name)
        if not field_id:
            if self.lenient_mode:
                logger.debug("No ClickUp field ID for '%s', skipping", field_name)
                continue
            raise ValueError(f"No ClickUp field ID configured for: {field_name}")

        clickup_value = self.convert_to_clickup_value(field_name, beads_value)
        if clickup_value is not None:
            clickup_fields.append(
                {
                    "id": field_id,
                    "value": clickup_value,
                }
            )

    return clickup_fields

clickup_custom_fields_to_beads(clickup_fields)

Convert ClickUp custom_fields array to beads fields dict.

Parameters:

Name	Type	Description	Default
clickup_fields	List[Dict[str, Any]]	List of ClickUp custom field dicts	required

Returns:

Type	Description
Dict[str, Any]	Dict of field_name -> value for beads

Source code in beads_clickup/field_mapper.py

def clickup_custom_fields_to_beads(
    self,
    clickup_fields: List[Dict[str, Any]],
) -> Dict[str, Any]:
    """Convert ClickUp custom_fields array to beads fields dict.

    Args:
        clickup_fields: List of ClickUp custom field dicts

    Returns:
        Dict of field_name -> value for beads
    """
    beads_fields = {}

    # Build reverse lookup: clickup_field_id -> beads_field_name
    id_to_name = {}
    for field_name, field_config in self.custom_fields.items():
        field_id = field_config.get("clickup_field_id")
        if field_id:
            id_to_name[field_id] = field_name

    for cu_field in clickup_fields:
        field_id = cu_field.get("id")
        field_name = id_to_name.get(field_id)

        if not field_name:
            # Try to use the ClickUp field name as fallback
            cu_name = cu_field.get("name", "").lower().replace(" ", "_")
            if cu_name in self.custom_fields:
                field_name = cu_name
            else:
                logger.debug("Unknown ClickUp field ID '%s', skipping", field_id)
                continue

        cu_value = cu_field.get("value")
        field_type = cu_field.get("type")
        beads_value = self.convert_from_clickup_value(
            field_name, cu_value, field_type
        )

        if beads_value is not None:
            beads_fields[field_name] = beads_value

    return beads_fields

convert_from_clickup_value(field_name, clickup_value, field_type=None)

Convert a ClickUp custom field value to beads format.

Parameters:

Name	Type	Description	Default
field_name	str	Beads field name	required
clickup_value	Any	Value from ClickUp task	required
field_type	Optional[str]	Override field type (useful when type is known from API)	None

Returns:

Type	Description
Optional[Any]	Value formatted for beads, or None if conversion fails

Source code in beads_clickup/field_mapper.py

def convert_from_clickup_value(
    self,
    field_name: str,
    clickup_value: Any,
    field_type: Optional[str] = None,
) -> Optional[Any]:
    """Convert a ClickUp custom field value to beads format.

    Args:
        field_name: Beads field name
        clickup_value: Value from ClickUp task
        field_type: Override field type (useful when type is known from API)

    Returns:
        Value formatted for beads, or None if conversion fails
    """
    if clickup_value is None:
        return None

    field_config = self.get_field_config(field_name)
    if not field_config and not field_type:
        if self.lenient_mode:
            logger.debug("Unknown custom field '%s', skipping", field_name)
            return None
        raise ValueError(f"Unknown custom field: {field_name}")

    ftype = field_type or (field_config.get("type") if field_config else "text")

    try:
        if ftype == "drop_down":
            return self._convert_dropdown_from_clickup(field_name, clickup_value)
        elif ftype in ("text", "short_text"):
            return str(clickup_value) if clickup_value else None
        elif ftype == "url":
            return str(clickup_value) if clickup_value else None
        elif ftype == "number":
            return float(clickup_value) if clickup_value else None
        elif ftype == "date":
            return int(clickup_value) if clickup_value else None
        elif ftype == "users":
            # ClickUp returns users as list of dicts with 'id' key
            if isinstance(clickup_value, list):
                return [
                    u.get("id") if isinstance(u, dict) else u for u in clickup_value
                ]
            return clickup_value
        elif ftype == "labels":
            if isinstance(clickup_value, list):
                return clickup_value
            return [clickup_value] if clickup_value else []
        else:
            return clickup_value

    except Exception as e:
        if self.lenient_mode:
            logger.warning(
                "Failed to convert ClickUp field '%s' value '%s': %s",
                field_name,
                clickup_value,
                e,
            )
            return None
        raise

convert_to_clickup_value(field_name, beads_value)

Convert a beads field value to ClickUp custom field format.

Parameters:

Name	Type	Description	Default
field_name	str	Beads field name	required
beads_value	Any	Value from beads issue	required

Returns:

Type	Description
Optional[Any]	Value formatted for ClickUp API, or None if conversion fails

Source code in beads_clickup/field_mapper.py

def convert_to_clickup_value(
    self,
    field_name: str,
    beads_value: Any,
) -> Optional[Any]:
    """Convert a beads field value to ClickUp custom field format.

    Args:
        field_name: Beads field name
        beads_value: Value from beads issue

    Returns:
        Value formatted for ClickUp API, or None if conversion fails
    """
    if beads_value is None:
        return None

    field_config = self.get_field_config(field_name)
    if not field_config:
        if self.lenient_mode:
            logger.debug("Unknown custom field '%s', skipping", field_name)
            return None
        raise ValueError(f"Unknown custom field: {field_name}")

    field_type = field_config.get("type", "text")

    try:
        if field_type == "drop_down":
            return self._convert_dropdown_to_clickup(field_name, beads_value)
        elif field_type in ("text", "short_text"):
            return str(beads_value)
        elif field_type == "url":
            return str(beads_value)
        elif field_type == "number":
            return float(beads_value)
        elif field_type == "date":
            # Expect Unix timestamp in milliseconds
            return int(beads_value)
        elif field_type == "users":
            # Expect user ID or list of user IDs
            if isinstance(beads_value, list):
                return beads_value
            return [beads_value]
        elif field_type == "labels":
            # Multi-select labels
            if isinstance(beads_value, list):
                return beads_value
            return [beads_value]
        else:
            logger.warning(
                "Unknown field type '%s' for field '%s'", field_type, field_name
            )
            return str(beads_value)

    except Exception as e:
        if self.lenient_mode:
            logger.warning(
                "Failed to convert field '%s' value '%s': %s",
                field_name,
                beads_value,
                e,
            )
            return None
        raise

get_clickup_field_id(field_name)

Get ClickUp field ID for a beads field name.

Handles both exact matches and normalized matches (strips numeric prefixes).

Parameters:

Name	Type	Description	Default
field_name	str	Beads field name	required

Returns:

Type	Description
Optional[str]	ClickUp custom field UUID or None

Source code in beads_clickup/field_mapper.py

def get_clickup_field_id(self, field_name: str) -> Optional[str]:
    """Get ClickUp field ID for a beads field name.

    Handles both exact matches and normalized matches (strips numeric prefixes).

    Args:
        field_name: Beads field name

    Returns:
        ClickUp custom field UUID or None
    """
    import re

    # Try exact match first
    field_config = self.get_field_config(field_name)
    if field_config:
        return field_config.get("clickup_field_id")

    # Try normalized name (strip numeric prefixes)
    normalized = re.sub(r'^\d+_', '', field_name)
    for config_name, config in self.custom_fields.items():
        config_normalized = re.sub(r'^\d+_', '', config_name)
        if config_normalized == normalized:
            return config.get("clickup_field_id")

    return None

get_field_config(field_name)

Get configuration for a custom field.

Handles both exact matches and normalized matches (strips numeric prefixes).

Parameters:

Name	Type	Description	Default
field_name	str	Beads field name	required

Returns:

Type	Description
Optional[Dict[str, Any]]	Field configuration dict or None if not configured

Source code in beads_clickup/field_mapper.py

def get_field_config(self, field_name: str) -> Optional[Dict[str, Any]]:
    """Get configuration for a custom field.

    Handles both exact matches and normalized matches (strips numeric prefixes).

    Args:
        field_name: Beads field name

    Returns:
        Field configuration dict or None if not configured
    """
    import re

    # Try exact match first
    if field_name in self.custom_fields:
        return self.custom_fields[field_name]

    # Try normalized name (strip numeric prefixes)
    normalized = re.sub(r'^\d+_', '', field_name)
    for config_name, config in self.custom_fields.items():
        config_normalized = re.sub(r'^\d+_', '', config_name)
        if config_normalized == normalized:
            return config

    return None

get_required_fields_with_defaults()

Get all required fields and their default values.

Returns:

Type	Description
Dict[str, Any]	Dict of field_name -> default_value for required fields

Source code in beads_clickup/field_mapper.py

def get_required_fields_with_defaults(self) -> Dict[str, Any]:
    """Get all required fields and their default values.

    Returns:
        Dict of field_name -> default_value for required fields
    """
    defaults = {}
    for field_name, field_config in self.custom_fields.items():
        if field_config.get("required", False):
            default = field_config.get("default")
            if default is not None:
                defaults[field_name] = default
    return defaults

set_dropdown_options(field_name, options)

Cache dropdown options for a field.

Parameters:

Name	Type	Description	Default
field_name	str	Beads field name (e.g., 'lifecycle_stage')	required
options	Dict[str, str]	Dict mapping option names to option IDs	required

Source code in beads_clickup/field_mapper.py

def set_dropdown_options(self, field_name: str, options: Dict[str, str]):
    """Cache dropdown options for a field.

    Args:
        field_name: Beads field name (e.g., 'lifecycle_stage')
        options: Dict mapping option names to option IDs
    """
    self._dropdown_options_cache[field_name] = options

validate_required_fields(beads_fields)

Validate and fill in required fields with defaults.

Parameters:

Name	Type	Description	Default
beads_fields	Dict[str, Any]	Dict of field_name -> value	required

Returns:

Type	Description
Dict[str, Any]	Updated dict with defaults filled in for missing required fields

Raises:

Type	Description
ValueError	If required field missing and no default (strict mode only)

Source code in beads_clickup/field_mapper.py

def validate_required_fields(
    self,
    beads_fields: Dict[str, Any],
) -> Dict[str, Any]:
    """Validate and fill in required fields with defaults.

    Args:
        beads_fields: Dict of field_name -> value

    Returns:
        Updated dict with defaults filled in for missing required fields

    Raises:
        ValueError: If required field missing and no default (strict mode only)
    """
    result = dict(beads_fields)

    for field_name, field_config in self.custom_fields.items():
        if not field_config.get("required", False):
            continue

        if field_name not in result or result[field_name] is None:
            default = field_config.get("default")
            if default is not None:
                result[field_name] = default
                logger.debug(
                    "Using default value '%s' for required field '%s'",
                    default,
                    field_name,
                )
            elif not self.lenient_mode:
                raise ValueError(
                    f"Required field '{field_name}' missing and no default configured"
                )

    return result

FieldMapper

Maps fields between beads issues and ClickUp tasks.

Source code in beads_clickup/field_mapper.py

class FieldMapper:
    """Maps fields between beads issues and ClickUp tasks."""

    def __init__(self, config: Dict[str, Any]):
        """Initialize field mapper with configuration.

        Args:
            config: Configuration dictionary from config.yaml
        """
        self.config = config
        self.field_mapping = config.get("field_mapping", {})
        self.custom_fields = self.field_mapping.get("custom_fields", {})
        self.status_map = self.field_mapping.get("status", {})
        self.priority_map = self.field_mapping.get("priority", {})
        self.use_external_ref = self.field_mapping.get("use_external_ref", True)
        self.labels_to_tags = self.field_mapping.get("labels_to_tags", True)

    def normalize_field_name(self, field_name: str) -> str:
        """Strip numeric prefixes from field names.

        Examples:
            01_lifecycle_stage -> lifecycle_stage
            02_category -> category
            lifecycle_stage -> lifecycle_stage (unchanged)

        Args:
            field_name: Field name, potentially with numeric prefix

        Returns:
            Normalized field name without numeric prefix
        """
        return re.sub(r'^\d+_', '', field_name)

    def get_field_config(self, field_name: str) -> Optional[Dict[str, Any]]:
        """Get configuration for a custom field.

        Args:
            field_name: Beads field name

        Returns:
            Field configuration dict or None if not configured
        """
        return self.custom_fields.get(field_name)

    def get_clickup_field_id(self, field_name: str) -> Optional[str]:
        """Get ClickUp field ID for a beads field name.

        Handles both exact matches and normalized matches (strips numeric prefixes).

        Args:
            field_name: Beads field name

        Returns:
            ClickUp custom field UUID or None
        """
        # Try exact match first
        field_config = self.get_field_config(field_name)
        if field_config:
            return field_config.get("clickup_field_id")

        # Try normalized name (strip numeric prefixes)
        normalized = self.normalize_field_name(field_name)
        for config_name, config in self.custom_fields.items():
            if self.normalize_field_name(config_name) == normalized:
                return config.get("clickup_field_id")

        return None

    def beads_to_clickup_status(self, beads_status: str) -> Optional[str]:
        """Map beads status to ClickUp status.

        Args:
            beads_status: Beads status (open, in_progress, completed, cancelled)

        Returns:
            ClickUp status name or None
        """
        return self.status_map.get(beads_status.lower())

    def clickup_to_beads_status(self, clickup_status: str) -> str:
        """Map ClickUp status to beads status.

        Uses explicit config mapping first, falls back to heuristics.

        Args:
            clickup_status: ClickUp status name (e.g., "UAT", "IN PROGRESS")

        Returns:
            Beads status (open, in_progress, completed, closed, cancelled)
        """
        # Explicit guard: PENDING REVIEW is a ClickUp-only status.
        # It must always map to in_progress in beads (never "pending_review").
        if clickup_status.strip().upper() == "PENDING REVIEW":
            return "in_progress"

        # Try explicit mapping first
        clickup_to_beads = self.status_map.get("clickup_to_beads", {})
        if clickup_status.lower() in clickup_to_beads:
            return clickup_to_beads[clickup_status.lower()]

        # Try reverse lookup in beads → ClickUp mapping
        clickup_lower = clickup_status.lower()
        for beads_status, cu_status in self.status_map.items():
            # Skip meta-keys added by build_status_config
            if beads_status in ("clickup_to_beads", "beads_categories"):
                continue
            if isinstance(cu_status, str) and cu_status.lower() == clickup_lower:
                return beads_status

        # Fall back to existing heuristics
        status_lower = clickup_status.lower()

        # Closed states
        if any(word in status_lower for word in ["complete", "done", "closed", "resolved", "finished"]):
            return "closed"

        # Cancelled states
        if any(word in status_lower for word in ["cancel", "cancelled", "abandoned"]):
            return "cancelled"

        # In progress states
        if any(word in status_lower for word in ["progress", "doing", "started", "active", "uat", "test", "review", "staging"]):
            return "in_progress"

        # Open states
        if any(word in status_lower for word in ["todo", "to do", "backlog", "open", "new"]):
            return "open"

        # Default to in_progress for unknown
        return "in_progress"

    def beads_to_clickup_priority(self, beads_priority) -> Optional[int]:
        """Map beads priority to ClickUp priority.

        Args:
            beads_priority: Beads priority (P0-P4, 0-4, or int)

        Returns:
            ClickUp priority (1-4) or None
        """
        # Convert to string if it's an integer
        beads_priority = str(beads_priority)

        # Normalize priority format
        if beads_priority.upper().startswith("P"):
            beads_priority = beads_priority[1:]  # Remove 'P' prefix

        priority_key = f"p{beads_priority.lower()}"
        return self.priority_map.get(priority_key)

    def clickup_to_beads_priority(self, clickup_priority: Optional[int]) -> str:
        """Map ClickUp priority to beads priority.

        Args:
            clickup_priority: ClickUp priority (1-4) or None

        Returns:
            Beads priority string (P0-P4)
        """
        if clickup_priority is None:
            return "2"  # Default to P2

        # Reverse lookup
        for beads_key, cu_pri in self.priority_map.items():
            if cu_pri == clickup_priority:
                # Extract number from p0, p1, etc.
                return beads_key[1]  # Return just the number

        # Fallback mapping
        priority_map_reverse = {
            1: "0",  # Urgent → P0
            2: "1",  # High → P1
            3: "2",  # Normal → P2
            4: "3",  # Low → P3
        }
        return priority_map_reverse.get(clickup_priority, "2")

    def beads_to_clickup_task(
        self,
        issue: Dict[str, Any],
        list_id: str,
        parent_task_id: Optional[str] = None,
        custom_field_mapper: Optional["CustomFieldMapper"] = None,
    ) -> Dict[str, Any]:
        """Convert a beads issue to ClickUp task creation arguments.

        Args:
            issue: Beads issue dictionary
            list_id: Target ClickUp list ID
            parent_task_id: Optional parent task ID (makes this a subtask)
            custom_field_mapper: Optional CustomFieldMapper for custom fields

        Returns:
            Dictionary of arguments for create_task()
        """
        from datetime import datetime, timedelta

        task_args = {
            "list_id": list_id,
            "name": issue.get("title", "Untitled"),
        }

        # Map description
        description = issue.get("description", "")
        if description:
            # Add beads issue ID to description for reference
            issue_id = issue.get("id", "")
            description += f"\n\n---\n*Synced from beads issue: {issue_id}*"
            task_args["description"] = description

        # Map status
        beads_status = issue.get("status", "open")
        clickup_status = self.beads_to_clickup_status(beads_status)
        if clickup_status:
            task_args["status"] = clickup_status

        # Map priority
        beads_priority = issue.get("priority", "2")
        clickup_priority = self.beads_to_clickup_priority(beads_priority)
        if clickup_priority is not None:
            task_args["priority"] = clickup_priority

        # Map labels to tags
        tags = []
        if self.labels_to_tags:
            labels = issue.get("labels", [])
            if labels:
                tags.extend(labels)

        # Always add beads-automation tag
        if "beads-automation" not in tags:
            tags.append("beads-automation")

        if tags:
            task_args["tags"] = tags

        # Set parent if provided
        if parent_task_id:
            task_args["parent"] = parent_task_id

        # Apply default standard fields from clickup.defaults config
        clickup_defaults = self.config.get("clickup", {}).get("defaults", {})

        # Default assignee
        default_assignee = clickup_defaults.get("assignee")
        if default_assignee:
            task_args["assignees"] = [str(default_assignee)]

        # Default time estimate
        default_time_estimate = clickup_defaults.get("time_estimate")
        if default_time_estimate:
            task_args["time_estimate"] = default_time_estimate

        # Auto start date (now)
        if clickup_defaults.get("auto_start_date", False):
            task_args["start_date"] = int(datetime.utcnow().timestamp() * 1000)

        # Auto due date (X days from now)
        auto_due_days = clickup_defaults.get("auto_due_date_days")
        if auto_due_days:
            due_date = datetime.utcnow() + timedelta(days=auto_due_days)
            task_args["due_date"] = int(due_date.timestamp() * 1000)

        # Build custom fields list
        custom_fields_list = []

        # Store beads issue ID in custom field (legacy approach)
        custom_field_name = self.field_mapping.get("clickup_custom_field_name")
        if custom_field_name and custom_field_mapper:
            bd_id_field_id = custom_field_mapper.get_clickup_field_id("bd_id")
            if bd_id_field_id:
                custom_fields_list.append(
                    {
                        "id": bd_id_field_id,
                        "value": issue.get("id", ""),
                    }
                )

        # Map custom fields using CustomFieldMapper
        if custom_field_mapper:
            # Use intelligent field inference
            from beads_clickup.field_inference import get_smart_defaults

            # Get required fields with defaults
            required_defaults = custom_field_mapper.get_required_fields_with_defaults()

            # Get inferred fields based on issue content
            inferred_fields = get_smart_defaults(issue, self.config)
            # Do not set lifecycle_stage from inference at create/sync; only from explicit issue or bd_assign_lifecycle
            inferred_fields.pop("lifecycle_stage", None)

            # Merge: config defaults < inferred < explicit issue fields
            issue_custom_fields = issue.get("custom_fields", {})
            merged_fields = {
                **required_defaults,
                **inferred_fields,
                **issue_custom_fields,
            }

            # Convert to ClickUp format
            cu_custom_fields = custom_field_mapper.beads_custom_fields_to_clickup(
                merged_fields
            )
            custom_fields_list.extend(cu_custom_fields)

        if custom_fields_list:
            task_args["custom_fields"] = custom_fields_list

        return task_args

    def clickup_to_beads_issue(
        self,
        task: Dict[str, Any],
    ) -> Dict[str, Any]:
        """Convert a ClickUp task to beads issue update fields.

        Args:
            task: ClickUp task dictionary

        Returns:
            Dictionary of beads issue fields to update
        """
        issue_update = {}

        # Map name/title
        if "name" in task:
            issue_update["title"] = task["name"]

        # Map description (strip beads reference if present)
        if "description" in task:
            description = task["description"]
            # Remove the "Synced from beads" footer if present
            if "---" in description and "Synced from beads issue:" in description:
                description = description.split("---")[0].strip()
            issue_update["description"] = description

        # Map status
        if "status" in task:
            # Handle both dict format and string format
            status_obj = task["status"]
            status_name = ""
            status_type = ""
            if isinstance(status_obj, dict):
                status_name = status_obj.get("status", "")
                status_type = status_obj.get("type", "")
            else:
                status_name = status_obj

            # ClickUp status type "closed" means task is completed/closed
            if status_type == "closed":
                issue_update["status"] = "completed"
            else:
                beads_status = self.clickup_to_beads_status(status_name)
                if beads_status:
                    issue_update["status"] = beads_status

        # Map priority
        if "priority" in task:
            priority = task.get("priority")
            # ClickUp returns priority as a dict with 'id' field
            if isinstance(priority, dict):
                priority = int(priority.get("id", 3)) if priority.get("id") else None
            beads_priority = self.clickup_to_beads_priority(priority)
            issue_update["priority"] = beads_priority

        # Map tags to labels
        if self.labels_to_tags and "tags" in task:
            tags = task.get("tags", [])
            if isinstance(tags, list):
                # Extract tag names if tags are dict objects
                label_names = []
                for tag in tags:
                    if isinstance(tag, dict):
                        label_names.append(tag.get("name", ""))
                    else:
                        label_names.append(str(tag))
                issue_update["labels"] = [l for l in label_names if l]

        # Extract beads issue ID from custom field if available
        custom_field_name = self.field_mapping.get("clickup_custom_field_name")
        if custom_field_name and "custom_fields" in task:
            custom_fields = task.get("custom_fields", [])
            for field in custom_fields:
                if field.get("name") == custom_field_name:
                    issue_update["_beads_id"] = field.get("value")
                    break

        return issue_update

    def should_sync_issue(self, issue: Dict[str, Any]) -> bool:
        """Determine if a beads issue should be synced to ClickUp.

        Args:
            issue: Beads issue dictionary

        Returns:
            True if issue should be synced
        """
        filters = self.config.get("filters", {})

        # Check sync_labels filter
        sync_labels = filters.get("sync_labels", [])
        if sync_labels:
            issue_labels = set(issue.get("labels", []))
            if not any(label in issue_labels for label in sync_labels):
                return False

        # Check exclude_labels filter
        exclude_labels = filters.get("exclude_labels", [])
        if exclude_labels:
            issue_labels = set(issue.get("labels", []))
            if any(label in issue_labels for label in exclude_labels):
                return False

        # Check sync_priorities filter
        sync_priorities = filters.get("sync_priorities", [])
        if sync_priorities:
            issue_priority = issue.get("priority", "")
            # Normalize priority format (P2 or 2 → p2)
            norm_priority = f"p{issue_priority.lower().replace('p', '')}"
            if norm_priority not in [p.lower() for p in sync_priorities]:
                return False

        # Check sync_types filter
        sync_types = filters.get("sync_types", [])
        if sync_types:
            issue_type = issue.get("type", "")
            if issue_type.lower() not in [t.lower() for t in sync_types]:
                return False

        return True

__init__(config)

Initialize field mapper with configuration.

Parameters:

Name	Type	Description	Default
config	Dict[str, Any]	Configuration dictionary from config.yaml	required

Source code in beads_clickup/field_mapper.py

def __init__(self, config: Dict[str, Any]):
    """Initialize field mapper with configuration.

    Args:
        config: Configuration dictionary from config.yaml
    """
    self.config = config
    self.field_mapping = config.get("field_mapping", {})
    self.custom_fields = self.field_mapping.get("custom_fields", {})
    self.status_map = self.field_mapping.get("status", {})
    self.priority_map = self.field_mapping.get("priority", {})
    self.use_external_ref = self.field_mapping.get("use_external_ref", True)
    self.labels_to_tags = self.field_mapping.get("labels_to_tags", True)

beads_to_clickup_priority(beads_priority)

Map beads priority to ClickUp priority.

Parameters:

Name	Type	Description	Default
beads_priority		Beads priority (P0-P4, 0-4, or int)	required

Returns:

Type	Description
Optional[int]	ClickUp priority (1-4) or None

Source code in beads_clickup/field_mapper.py

def beads_to_clickup_priority(self, beads_priority) -> Optional[int]:
    """Map beads priority to ClickUp priority.

    Args:
        beads_priority: Beads priority (P0-P4, 0-4, or int)

    Returns:
        ClickUp priority (1-4) or None
    """
    # Convert to string if it's an integer
    beads_priority = str(beads_priority)

    # Normalize priority format
    if beads_priority.upper().startswith("P"):
        beads_priority = beads_priority[1:]  # Remove 'P' prefix

    priority_key = f"p{beads_priority.lower()}"
    return self.priority_map.get(priority_key)

beads_to_clickup_status(beads_status)

Map beads status to ClickUp status.

Parameters:

Name	Type	Description	Default
beads_status	str	Beads status (open, in_progress, completed, cancelled)	required

Returns:

Type	Description
Optional[str]	ClickUp status name or None

Source code in beads_clickup/field_mapper.py

def beads_to_clickup_status(self, beads_status: str) -> Optional[str]:
    """Map beads status to ClickUp status.

    Args:
        beads_status: Beads status (open, in_progress, completed, cancelled)

    Returns:
        ClickUp status name or None
    """
    return self.status_map.get(beads_status.lower())

beads_to_clickup_task(issue, list_id, parent_task_id=None, custom_field_mapper=None)

Convert a beads issue to ClickUp task creation arguments.

Parameters:

Name	Type	Description	Default
issue	Dict[str, Any]	Beads issue dictionary	required
list_id	str	Target ClickUp list ID	required
parent_task_id	Optional[str]	Optional parent task ID (makes this a subtask)	None
custom_field_mapper	Optional[CustomFieldMapper]	Optional CustomFieldMapper for custom fields	None

Returns:

Type	Description
Dict[str, Any]	Dictionary of arguments for create_task()

Source code in beads_clickup/field_mapper.py

def beads_to_clickup_task(
    self,
    issue: Dict[str, Any],
    list_id: str,
    parent_task_id: Optional[str] = None,
    custom_field_mapper: Optional["CustomFieldMapper"] = None,
) -> Dict[str, Any]:
    """Convert a beads issue to ClickUp task creation arguments.

    Args:
        issue: Beads issue dictionary
        list_id: Target ClickUp list ID
        parent_task_id: Optional parent task ID (makes this a subtask)
        custom_field_mapper: Optional CustomFieldMapper for custom fields

    Returns:
        Dictionary of arguments for create_task()
    """
    from datetime import datetime, timedelta

    task_args = {
        "list_id": list_id,
        "name": issue.get("title", "Untitled"),
    }

    # Map description
    description = issue.get("description", "")
    if description:
        # Add beads issue ID to description for reference
        issue_id = issue.get("id", "")
        description += f"\n\n---\n*Synced from beads issue: {issue_id}*"
        task_args["description"] = description

    # Map status
    beads_status = issue.get("status", "open")
    clickup_status = self.beads_to_clickup_status(beads_status)
    if clickup_status:
        task_args["status"] = clickup_status

    # Map priority
    beads_priority = issue.get("priority", "2")
    clickup_priority = self.beads_to_clickup_priority(beads_priority)
    if clickup_priority is not None:
        task_args["priority"] = clickup_priority

    # Map labels to tags
    tags = []
    if self.labels_to_tags:
        labels = issue.get("labels", [])
        if labels:
            tags.extend(labels)

    # Always add beads-automation tag
    if "beads-automation" not in tags:
        tags.append("beads-automation")

    if tags:
        task_args["tags"] = tags

    # Set parent if provided
    if parent_task_id:
        task_args["parent"] = parent_task_id

    # Apply default standard fields from clickup.defaults config
    clickup_defaults = self.config.get("clickup", {}).get("defaults", {})

    # Default assignee
    default_assignee = clickup_defaults.get("assignee")
    if default_assignee:
        task_args["assignees"] = [str(default_assignee)]

    # Default time estimate
    default_time_estimate = clickup_defaults.get("time_estimate")
    if default_time_estimate:
        task_args["time_estimate"] = default_time_estimate

    # Auto start date (now)
    if clickup_defaults.get("auto_start_date", False):
        task_args["start_date"] = int(datetime.utcnow().timestamp() * 1000)

    # Auto due date (X days from now)
    auto_due_days = clickup_defaults.get("auto_due_date_days")
    if auto_due_days:
        due_date = datetime.utcnow() + timedelta(days=auto_due_days)
        task_args["due_date"] = int(due_date.timestamp() * 1000)

    # Build custom fields list
    custom_fields_list = []

    # Store beads issue ID in custom field (legacy approach)
    custom_field_name = self.field_mapping.get("clickup_custom_field_name")
    if custom_field_name and custom_field_mapper:
        bd_id_field_id = custom_field_mapper.get_clickup_field_id("bd_id")
        if bd_id_field_id:
            custom_fields_list.append(
                {
                    "id": bd_id_field_id,
                    "value": issue.get("id", ""),
                }
            )

    # Map custom fields using CustomFieldMapper
    if custom_field_mapper:
        # Use intelligent field inference
        from beads_clickup.field_inference import get_smart_defaults

        # Get required fields with defaults
        required_defaults = custom_field_mapper.get_required_fields_with_defaults()

        # Get inferred fields based on issue content
        inferred_fields = get_smart_defaults(issue, self.config)
        # Do not set lifecycle_stage from inference at create/sync; only from explicit issue or bd_assign_lifecycle
        inferred_fields.pop("lifecycle_stage", None)

        # Merge: config defaults < inferred < explicit issue fields
        issue_custom_fields = issue.get("custom_fields", {})
        merged_fields = {
            **required_defaults,
            **inferred_fields,
            **issue_custom_fields,
        }

        # Convert to ClickUp format
        cu_custom_fields = custom_field_mapper.beads_custom_fields_to_clickup(
            merged_fields
        )
        custom_fields_list.extend(cu_custom_fields)

    if custom_fields_list:
        task_args["custom_fields"] = custom_fields_list

    return task_args

clickup_to_beads_issue(task)

Convert a ClickUp task to beads issue update fields.

Parameters:

Name	Type	Description	Default
task	Dict[str, Any]	ClickUp task dictionary	required

Returns:

Type	Description
Dict[str, Any]	Dictionary of beads issue fields to update

Source code in beads_clickup/field_mapper.py

def clickup_to_beads_issue(
    self,
    task: Dict[str, Any],
) -> Dict[str, Any]:
    """Convert a ClickUp task to beads issue update fields.

    Args:
        task: ClickUp task dictionary

    Returns:
        Dictionary of beads issue fields to update
    """
    issue_update = {}

    # Map name/title
    if "name" in task:
        issue_update["title"] = task["name"]

    # Map description (strip beads reference if present)
    if "description" in task:
        description = task["description"]
        # Remove the "Synced from beads" footer if present
        if "---" in description and "Synced from beads issue:" in description:
            description = description.split("---")[0].strip()
        issue_update["description"] = description

    # Map status
    if "status" in task:
        # Handle both dict format and string format
        status_obj = task["status"]
        status_name = ""
        status_type = ""
        if isinstance(status_obj, dict):
            status_name = status_obj.get("status", "")
            status_type = status_obj.get("type", "")
        else:
            status_name = status_obj

        # ClickUp status type "closed" means task is completed/closed
        if status_type == "closed":
            issue_update["status"] = "completed"
        else:
            beads_status = self.clickup_to_beads_status(status_name)
            if beads_status:
                issue_update["status"] = beads_status

    # Map priority
    if "priority" in task:
        priority = task.get("priority")
        # ClickUp returns priority as a dict with 'id' field
        if isinstance(priority, dict):
            priority = int(priority.get("id", 3)) if priority.get("id") else None
        beads_priority = self.clickup_to_beads_priority(priority)
        issue_update["priority"] = beads_priority

    # Map tags to labels
    if self.labels_to_tags and "tags" in task:
        tags = task.get("tags", [])
        if isinstance(tags, list):
            # Extract tag names if tags are dict objects
            label_names = []
            for tag in tags:
                if isinstance(tag, dict):
                    label_names.append(tag.get("name", ""))
                else:
                    label_names.append(str(tag))
            issue_update["labels"] = [l for l in label_names if l]

    # Extract beads issue ID from custom field if available
    custom_field_name = self.field_mapping.get("clickup_custom_field_name")
    if custom_field_name and "custom_fields" in task:
        custom_fields = task.get("custom_fields", [])
        for field in custom_fields:
            if field.get("name") == custom_field_name:
                issue_update["_beads_id"] = field.get("value")
                break

    return issue_update

clickup_to_beads_priority(clickup_priority)

Map ClickUp priority to beads priority.

Parameters:

Name	Type	Description	Default
clickup_priority	Optional[int]	ClickUp priority (1-4) or None	required

Returns:

Type	Description
str	Beads priority string (P0-P4)

Source code in beads_clickup/field_mapper.py

def clickup_to_beads_priority(self, clickup_priority: Optional[int]) -> str:
    """Map ClickUp priority to beads priority.

    Args:
        clickup_priority: ClickUp priority (1-4) or None

    Returns:
        Beads priority string (P0-P4)
    """
    if clickup_priority is None:
        return "2"  # Default to P2

    # Reverse lookup
    for beads_key, cu_pri in self.priority_map.items():
        if cu_pri == clickup_priority:
            # Extract number from p0, p1, etc.
            return beads_key[1]  # Return just the number

    # Fallback mapping
    priority_map_reverse = {
        1: "0",  # Urgent → P0
        2: "1",  # High → P1
        3: "2",  # Normal → P2
        4: "3",  # Low → P3
    }
    return priority_map_reverse.get(clickup_priority, "2")

clickup_to_beads_status(clickup_status)

Map ClickUp status to beads status.

Uses explicit config mapping first, falls back to heuristics.

Parameters:

Name	Type	Description	Default
clickup_status	str	ClickUp status name (e.g., "UAT", "IN PROGRESS")	required

Returns:

Type	Description
str	Beads status (open, in_progress, completed, closed, cancelled)

Source code in beads_clickup/field_mapper.py

def clickup_to_beads_status(self, clickup_status: str) -> str:
    """Map ClickUp status to beads status.

    Uses explicit config mapping first, falls back to heuristics.

    Args:
        clickup_status: ClickUp status name (e.g., "UAT", "IN PROGRESS")

    Returns:
        Beads status (open, in_progress, completed, closed, cancelled)
    """
    # Explicit guard: PENDING REVIEW is a ClickUp-only status.
    # It must always map to in_progress in beads (never "pending_review").
    if clickup_status.strip().upper() == "PENDING REVIEW":
        return "in_progress"

    # Try explicit mapping first
    clickup_to_beads = self.status_map.get("clickup_to_beads", {})
    if clickup_status.lower() in clickup_to_beads:
        return clickup_to_beads[clickup_status.lower()]

    # Try reverse lookup in beads → ClickUp mapping
    clickup_lower = clickup_status.lower()
    for beads_status, cu_status in self.status_map.items():
        # Skip meta-keys added by build_status_config
        if beads_status in ("clickup_to_beads", "beads_categories"):
            continue
        if isinstance(cu_status, str) and cu_status.lower() == clickup_lower:
            return beads_status

    # Fall back to existing heuristics
    status_lower = clickup_status.lower()

    # Closed states
    if any(word in status_lower for word in ["complete", "done", "closed", "resolved", "finished"]):
        return "closed"

    # Cancelled states
    if any(word in status_lower for word in ["cancel", "cancelled", "abandoned"]):
        return "cancelled"

    # In progress states
    if any(word in status_lower for word in ["progress", "doing", "started", "active", "uat", "test", "review", "staging"]):
        return "in_progress"

    # Open states
    if any(word in status_lower for word in ["todo", "to do", "backlog", "open", "new"]):
        return "open"

    # Default to in_progress for unknown
    return "in_progress"

get_clickup_field_id(field_name)

Get ClickUp field ID for a beads field name.

Handles both exact matches and normalized matches (strips numeric prefixes).

Parameters:

Name	Type	Description	Default
field_name	str	Beads field name	required

Returns:

Type	Description
Optional[str]	ClickUp custom field UUID or None

Source code in beads_clickup/field_mapper.py

def get_clickup_field_id(self, field_name: str) -> Optional[str]:
    """Get ClickUp field ID for a beads field name.

    Handles both exact matches and normalized matches (strips numeric prefixes).

    Args:
        field_name: Beads field name

    Returns:
        ClickUp custom field UUID or None
    """
    # Try exact match first
    field_config = self.get_field_config(field_name)
    if field_config:
        return field_config.get("clickup_field_id")

    # Try normalized name (strip numeric prefixes)
    normalized = self.normalize_field_name(field_name)
    for config_name, config in self.custom_fields.items():
        if self.normalize_field_name(config_name) == normalized:
            return config.get("clickup_field_id")

    return None

get_field_config(field_name)

Get configuration for a custom field.

Parameters:

Name	Type	Description	Default
field_name	str	Beads field name	required

Returns:

Type	Description
Optional[Dict[str, Any]]	Field configuration dict or None if not configured

Source code in beads_clickup/field_mapper.py

def get_field_config(self, field_name: str) -> Optional[Dict[str, Any]]:
    """Get configuration for a custom field.

    Args:
        field_name: Beads field name

    Returns:
        Field configuration dict or None if not configured
    """
    return self.custom_fields.get(field_name)

normalize_field_name(field_name)

Strip numeric prefixes from field names.

Examples:

01_lifecycle_stage -> lifecycle_stage 02_category -> category lifecycle_stage -> lifecycle_stage (unchanged)

Parameters:

Name	Type	Description	Default
field_name	str	Field name, potentially with numeric prefix	required

Returns:

Type	Description
str	Normalized field name without numeric prefix

Source code in beads_clickup/field_mapper.py

def normalize_field_name(self, field_name: str) -> str:
    """Strip numeric prefixes from field names.

    Examples:
        01_lifecycle_stage -> lifecycle_stage
        02_category -> category
        lifecycle_stage -> lifecycle_stage (unchanged)

    Args:
        field_name: Field name, potentially with numeric prefix

    Returns:
        Normalized field name without numeric prefix
    """
    return re.sub(r'^\d+_', '', field_name)

should_sync_issue(issue)

Determine if a beads issue should be synced to ClickUp.

Parameters:

Name	Type	Description	Default
issue	Dict[str, Any]	Beads issue dictionary	required

Returns:

Type	Description
bool	True if issue should be synced

Source code in beads_clickup/field_mapper.py

def should_sync_issue(self, issue: Dict[str, Any]) -> bool:
    """Determine if a beads issue should be synced to ClickUp.

    Args:
        issue: Beads issue dictionary

    Returns:
        True if issue should be synced
    """
    filters = self.config.get("filters", {})

    # Check sync_labels filter
    sync_labels = filters.get("sync_labels", [])
    if sync_labels:
        issue_labels = set(issue.get("labels", []))
        if not any(label in issue_labels for label in sync_labels):
            return False

    # Check exclude_labels filter
    exclude_labels = filters.get("exclude_labels", [])
    if exclude_labels:
        issue_labels = set(issue.get("labels", []))
        if any(label in issue_labels for label in exclude_labels):
            return False

    # Check sync_priorities filter
    sync_priorities = filters.get("sync_priorities", [])
    if sync_priorities:
        issue_priority = issue.get("priority", "")
        # Normalize priority format (P2 or 2 → p2)
        norm_priority = f"p{issue_priority.lower().replace('p', '')}"
        if norm_priority not in [p.lower() for p in sync_priorities]:
            return False

    # Check sync_types filter
    sync_types = filters.get("sync_types", [])
    if sync_types:
        issue_type = issue.get("type", "")
        if issue_type.lower() not in [t.lower() for t in sync_types]:
            return False

    return True