"""Write executors for label operations: create_label, update_label, delete_label. All executors open their own DB session via ``AsyncSessionLocal`` so they can be called from the MCP dispatcher without sharing a session with unrelated requests. Input validation mirrors the REST API constraints (same field limits and patterns) so agents get a clean error before any SQL is executed. """ import logging import re from datetime import datetime, timezone from musehub.core.genesis import compute_label_id from musehub.types.json_types import JSONObject from musehub.db.database import AsyncSessionLocal from musehub.services import musehub_repository from musehub.services.musehub_mcp_executor import MusehubToolResult, _check_db_available from musehub.mcp.write_tools.issues import _require_write_access logger = logging.getLogger(__name__) # Mirrors REST API constraints — keep in sync with labels.py. _MAX_LABEL_NAME = 50 _MAX_LABEL_DESC = 200 _HEX_COLOR_RE = re.compile(r"^#[0-9a-fA-F]{6}$") def _validate_color(color: str) -> bool: """Return True iff *color* is a valid 7-char hex colour string (e.g. '#d73a4a').""" return bool(_HEX_COLOR_RE.match(color)) async def execute_create_label( *, repo_id: str, name: str, color: str, description: str = "", actor: str = "", ) -> MusehubToolResult: """Create a repo-scoped label with a name and hex colour. Label names must be unique within the repository. The ``color`` must be a 7-character hex string starting with ``#`` (e.g. ``'#d73a4a'``). Pass ``description`` to attach a human-readable explanation. Args: repo_id: sha256 genesis ID of the repository. name: Label name (unique per repo, ≤ 50 chars). color: 7-char hex colour string starting with ``#`` (e.g. ``'#d73a4a'``). description: Optional label description (≤ 200 chars). actor: Authenticated user handle (unused for DB write; logged for audit). Returns: ``MusehubToolResult`` with ``data.label_id`` on success. """ if (err := _check_db_available()) is not None: return err # ── Local validation — fail before touching the DB ──────────────────────── name = name.strip() if not name: return MusehubToolResult( ok=False, error_code="invalid_args", error_message="Label name must not be empty.", ) if len(name) > _MAX_LABEL_NAME: return MusehubToolResult( ok=False, error_code="invalid_args", error_message=f"Label name is too long ({len(name)} chars); maximum is {_MAX_LABEL_NAME}.", ) color = color.strip() if not _validate_color(color): return MusehubToolResult( ok=False, error_code="invalid_args", error_message=( f"Invalid colour '{color}'. " "Must be a 7-character hex string starting with '#' (e.g. '#d73a4a')." ), ) if len(description) > _MAX_LABEL_DESC: return MusehubToolResult( ok=False, error_code="invalid_args", error_message=f"Description is too long ({len(description)} chars); maximum is {_MAX_LABEL_DESC}.", ) try: from sqlalchemy import text async with AsyncSessionLocal() as session: repo = await musehub_repository.get_repo(session, repo_id) if repo is None: return MusehubToolResult( ok=False, error_code="repo_not_found", error_message=f"Repository '{repo_id}' not found.", hint="Call musehub_search_repos() to find available repositories.", ) if (err := await _require_write_access(session, repo_id, actor, repo.owner)) is not None: return err existing = await session.execute( text( "SELECT 1 FROM musehub_labels " "WHERE repo_id = :repo_id AND name = :name" ), {"repo_id": repo_id, "name": name}, ) if existing.scalar_one_or_none() is not None: return MusehubToolResult( ok=False, error_code="already_exists", error_message=f"Label '{name}' already exists in repository '{repo_id}'.", ) _now = datetime.now(timezone.utc) label_id = compute_label_id(repo_id, name, _now.isoformat()) await session.execute( text( "INSERT INTO musehub_labels " "(id, repo_id, name, color, description, created_at) " "VALUES (:label_id, :repo_id, :name, :color, :description, :created_at)" ), { "label_id": label_id, "repo_id": repo_id, "name": name, "color": color, "description": description, "created_at": _now, }, ) await session.commit() data: JSONObject = { "label_id": label_id, "repo_id": repo_id, "name": name, "color": color, "description": description, } logger.info("MCP create_label '%s' (%s) in repo %s", name, label_id, repo_id) return MusehubToolResult(ok=True, data=data) except Exception as exc: logger.exception("MCP create_label failed: %s", exc) return MusehubToolResult( ok=False, error_code="internal_error", error_message=str(exc), ) async def execute_update_label( *, repo_id: str, label_id: str, name: str | None = None, color: str | None = None, description: str | None = None, actor: str = "", ) -> MusehubToolResult: """Partially update an existing label. Only supplied fields are changed; omitted fields are left unchanged. Pass ``name`` to rename the label (still must be unique within the repo). Args: repo_id: sha256 genesis ID of the repository. label_id: ID of the label to update. name: New label name (optional, ≤ 50 chars). color: New hex colour (optional, e.g. ``'#b60205'``). description: New description (optional, ≤ 200 chars; pass ``""`` to clear). actor: Authenticated user handle (logged for audit). Returns: ``MusehubToolResult`` with the updated label data on success. """ if (err := _check_db_available()) is not None: return err if name is None and color is None and description is None: return MusehubToolResult( ok=False, error_code="invalid_args", error_message="Provide at least one of: name, color, description.", ) # ── Local validation ────────────────────────────────────────────────────── if name is not None: name = name.strip() if not name: return MusehubToolResult( ok=False, error_code="invalid_args", error_message="New label name must not be empty.", ) if len(name) > _MAX_LABEL_NAME: return MusehubToolResult( ok=False, error_code="invalid_args", error_message=f"New name is too long ({len(name)} chars); maximum is {_MAX_LABEL_NAME}.", ) if color is not None: color = color.strip() if not _validate_color(color): return MusehubToolResult( ok=False, error_code="invalid_args", error_message=( f"Invalid colour '{color}'. " "Must be a 7-character hex string starting with '#'." ), ) if description is not None and len(description) > _MAX_LABEL_DESC: return MusehubToolResult( ok=False, error_code="invalid_args", error_message=f"Description is too long ({len(description)} chars); maximum is {_MAX_LABEL_DESC}.", ) try: from sqlalchemy import text async with AsyncSessionLocal() as session: repo = await musehub_repository.get_repo(session, repo_id) if repo is None: return MusehubToolResult( ok=False, error_code="repo_not_found", error_message=f"Repository '{repo_id}' not found.", hint="Call musehub_search_repos() to find available repositories.", ) if (err := await _require_write_access(session, repo_id, actor, repo.owner)) is not None: return err # Fetch existing label. row = await session.execute( text( "SELECT id AS label_id, repo_id, name, color, description " "FROM musehub_labels " "WHERE id = :label_id AND repo_id = :repo_id" ), {"label_id": label_id, "repo_id": repo_id}, ) existing = row.mappings().one_or_none() if existing is None: return MusehubToolResult( ok=False, error_code="not_found", error_message=f"Label '{label_id}' not found in repository '{repo_id}'.", hint="Call musehub_list_labels() to list available labels.", ) new_name = name if name is not None else existing["name"] new_color = color if color is not None else existing["color"] new_description = description if description is not None else existing["description"] # Name uniqueness check when renaming. if name is not None and name != existing["name"]: dupe = await session.execute( text( "SELECT 1 FROM musehub_labels " "WHERE repo_id = :repo_id AND name = :name AND id != :label_id" ), {"repo_id": repo_id, "name": name, "label_id": label_id}, ) if dupe.scalar_one_or_none() is not None: return MusehubToolResult( ok=False, error_code="already_exists", error_message=f"Label '{name}' already exists in repository '{repo_id}'.", ) await session.execute( text( "UPDATE musehub_labels " "SET name = :name, color = :color, description = :description " "WHERE id = :label_id AND repo_id = :repo_id" ), { "name": new_name, "color": new_color, "description": new_description, "label_id": label_id, "repo_id": repo_id, }, ) await session.commit() data: JSONObject = { "label_id": label_id, "repo_id": repo_id, "name": new_name, "color": new_color, "description": new_description, } logger.info("MCP update_label %s in repo %s", label_id, repo_id) return MusehubToolResult(ok=True, data=data) except Exception as exc: logger.exception("MCP update_label failed: %s", exc) return MusehubToolResult( ok=False, error_code="internal_error", error_message=str(exc), ) async def execute_delete_label( *, repo_id: str, label_id: str, actor: str = "", ) -> MusehubToolResult: """Permanently delete a label and remove it from all issues and proposals. Args: repo_id: sha256 genesis ID of the repository. label_id: ID of the label to delete. actor: Authenticated user handle (logged for audit). Returns: ``MusehubToolResult`` with ``data.deleted`` on success. """ if (err := _check_db_available()) is not None: return err try: from sqlalchemy import text async with AsyncSessionLocal() as session: repo = await musehub_repository.get_repo(session, repo_id) if repo is None: return MusehubToolResult( ok=False, error_code="repo_not_found", error_message=f"Repository '{repo_id}' not found.", hint="Call musehub_search_repos() to find available repositories.", ) if (err := await _require_write_access(session, repo_id, actor, repo.owner)) is not None: return err row = await session.execute( text( "SELECT id, name FROM musehub_labels " "WHERE id = :label_id AND repo_id = :repo_id" ), {"label_id": label_id, "repo_id": repo_id}, ) existing = row.mappings().one_or_none() if existing is None: return MusehubToolResult( ok=False, error_code="not_found", error_message=f"Label '{label_id}' not found in repository '{repo_id}'.", hint="Call musehub_list_labels() to list available labels.", ) label_name: str = existing["name"] # Remove proposal associations before deleting the label row. await session.execute( text("DELETE FROM musehub_proposal_labels WHERE label_id = :label_id"), {"label_id": label_id}, ) await session.execute( text("DELETE FROM musehub_labels WHERE id = :label_id AND repo_id = :repo_id"), {"label_id": label_id, "repo_id": repo_id}, ) await session.commit() data: JSONObject = { "deleted": True, "label_id": label_id, "repo_id": repo_id, "name": label_name, } logger.info("MCP delete_label %s ('%s') from repo %s", label_id, label_name, repo_id) return MusehubToolResult(ok=True, data=data) except Exception as exc: logger.exception("MCP delete_label failed: %s", exc) return MusehubToolResult( ok=False, error_code="internal_error", error_message=str(exc), )