bb_ops.py

"""
bb_ops — Bitbucket REST operations grouped by resource.

The MCP server (mcp_server.py, future PR) wires each function here to a
tool. Each function takes a BBClient as its first positional argument and
returns native Python data (dicts, lists, strings) — no terminal-style
formatting, no colour codes, no parsing of bash output.

`bb` (bash) and bb_ops (Python) are parallel implementations of the same
Bitbucket REST contract. See CONTRIBUTING.md for the parity rule: when a
defect surfaces in either side, the fix lands in both code paths.

Current scope: pipelines, pull requests, repos/branches/vars/downloads/
commits. The companion git_ops module provides the git-context wrappers
the MCP server uses to resolve "current branch" / "remote workspace"
before invoking these ops.
"""

from __future__ import annotations

from typing import Any, Iterable
from urllib.parse import quote

from bb_api import BBApiError, BBClient, repo_path


# ---------------------------------------------------------------------------
# Internal helpers
# ---------------------------------------------------------------------------

# Bitbucket Cloud caps pagelen at 100 server-side. Asking for more is
# silently truncated, which would create a confusing partial result. Clamp
# explicitly so the caller's intent (cap N items, paginate if needed) is
# preserved.
_BITBUCKET_MAX_PAGELEN = 100


def _is_positive_int(value: Any) -> bool:
    """True iff `value` is an int (NOT a bool) >= 1.

    `bool` is a subclass of `int` in Python, so `isinstance(True, int)` is
    True and `True < 1` is False — meaning a bare `isinstance(x, int) and
    x >= 1` check happily accepts `True` as `1`. That then propagates
    through f-string interpolation into URLs as the literal `"True"`, and
    through `urlencode({"pagelen": True})` as `"pagelen=True"` — both
    failure modes the boundary validator exists to prevent.
    """
    return isinstance(value, int) and not isinstance(value, bool) and value >= 1

# When resolving a build_number -> uuid, we walk the pipelines list sorted
# by most-recent-first. This cap bounds how far back we look before giving
# up. 2000 = 20 pages of 100 = "any pipeline triggered in the last few
# months" for an active repo. The bash script's inline lookup is a single
# 100-pipeline page; this MCP-side scan trades a few extra API calls for
# the ability to address older builds by number.
_PIPELINE_SCAN_LIMIT = 2000


class BBOpNotFound(LookupError):
    """A requested resource (pipeline build_number, step index, etc.) was
    not present in the responses we walked. Distinct from BBApiError so
    callers can render "no such pipeline" vs "API failure" differently."""


def _wrap_uuid(uuid: str) -> str:
    """Bitbucket's URL contract uses `{uuid}` with the literal braces,
    URL-encoded as `%7B...%7D`. The bash script does this by interpolating
    `%7B${uuid}%7D` directly into curl URLs; mirror that. The UUID itself
    is alphanumeric+hyphens so `quote` would no-op on it, but we route
    through it to defend against a UUID that ever contains characters
    that would otherwise need encoding."""
    inner = uuid.strip()
    if inner.startswith("{") and inner.endswith("}"):
        inner = inner[1:-1]
    return f"%7B{quote(inner)}%7D"


def _pipelines_root(workspace: str, repo: str) -> str:
    """Common URL prefix for the pipelines API. Centralising this means a
    future API-version bump touches one place."""
    return f"{repo_path(workspace, repo)}/pipelines/"


def _strip_uuid_braces(uuid: str | None) -> str:
    """Bitbucket returns pipeline UUIDs in two shapes depending on endpoint:
    bare ('a1b2-...') and brace-wrapped ('{a1b2-...}'). Normalise to bare
    so callers don't have to care."""
    if not uuid:
        raise BBApiError(0, "", "response missing uuid")
    s = uuid.strip()
    if s.startswith("{") and s.endswith("}"):
        return s[1:-1]
    return s


# ---------------------------------------------------------------------------
# Resolution helpers (build_number -> uuid, step_index -> uuid)
# ---------------------------------------------------------------------------


def _resolve_pipeline_uuid(
    client: BBClient,
    workspace: str,
    repo: str,
    build_number: int,
    *,
    scan_limit: int = _PIPELINE_SCAN_LIMIT,
) -> str:
    """Resolve a pipeline's UUID by walking pipelines/ sorted by most-recent.

    Bitbucket Cloud's API does not expose a direct `GET /pipelines/{build_number}`
    endpoint, only `GET /pipelines/{uuid}`. The CLI passes a build_number
    because that's what's user-visible (and what the API echoes in payloads).
    This helper paginates the listing until it finds the matching build or
    `scan_limit` items have been examined.

    Raises BBOpNotFound if the build_number isn't found within the scan
    window. Distinct from a network/API failure so callers can render
    "no such pipeline #N" naturally.
    """
    if not _is_positive_int(build_number):
        raise ValueError(f"build_number must be a positive int, got {build_number!r}")

    seen = 0
    path = _pipelines_root(workspace, repo)
    query = {"sort": "-created_on", "pagelen": _BITBUCKET_MAX_PAGELEN}
    for pipeline in client.paginate(path, query=query):
        seen += 1
        if pipeline.get("build_number") == build_number:
            return _strip_uuid_braces(pipeline.get("uuid"))
        if seen >= scan_limit:
            break
    raise BBOpNotFound(
        f"pipeline #{build_number} not found within the {scan_limit} most-recent "
        f"pipelines of {workspace}/{repo}"
    )


def _resolve_step_uuid(
    client: BBClient,
    workspace: str,
    repo: str,
    pipeline_uuid: str,
    step_index: int,
) -> str:
    """Return the step UUID for the step at the given 0-based index.

    The bash script's `bb logs` uses the same 0-based indexing into the
    steps list; mirror that contract so the user-facing index numbers
    match across both surfaces.

    Returns just the uuid (not the name) — callers that need the name
    should fetch the steps list themselves via `pipeline_steps()`. The
    MCP step-logs tool wraps the log payload in its own response shape
    and can surface the name there.
    """
    if (
        not isinstance(step_index, int)
        or isinstance(step_index, bool)
        or step_index < 0
    ):
        raise ValueError(f"step_index must be a non-negative int, got {step_index!r}")

    steps = _pipeline_steps_by_uuid(client, workspace, repo, pipeline_uuid)
    if step_index >= len(steps):
        raise BBOpNotFound(
            f"step index {step_index} out of range "
            f"(pipeline has {len(steps)} step{'s' if len(steps) != 1 else ''})"
        )
    return _strip_uuid_braces(steps[step_index].get("uuid"))


# ---------------------------------------------------------------------------
# Public operations
# ---------------------------------------------------------------------------


def pipelines_list(
    client: BBClient,
    workspace: str,
    repo: str,
    *,
    count: int = 10,
    sort: str = "-created_on",
    branch: str | None = None,
) -> list[dict[str, Any]]:
    """List recent pipelines, most-recent first by default.

    `count` is the upper bound on returned items. We always honour it
    even if it exceeds Bitbucket's per-page cap (100): the function
    paginates as needed.

    `branch` filters to pipelines triggered against a specific branch via
    Bitbucket's `target.ref_name` query (the API supports this without a
    `?q=` filter shape).
    """
    if not _is_positive_int(count):
        raise ValueError(f"count must be a positive int, got {count!r}")

    pagelen = min(count, _BITBUCKET_MAX_PAGELEN)
    query: dict[str, Any] = {"sort": sort, "pagelen": pagelen}
    if branch is not None:
        query["target.ref_name"] = branch

    out: list[dict[str, Any]] = []
    for pipeline in client.paginate(_pipelines_root(workspace, repo), query=query):
        out.append(pipeline)
        if len(out) >= count:
            break
    return out


def pipeline_show(
    client: BBClient, workspace: str, repo: str, build_number: int
) -> dict[str, Any]:
    """Fetch full pipeline detail for the given build_number."""
    uuid = _resolve_pipeline_uuid(client, workspace, repo, build_number)
    return client.get(f"{_pipelines_root(workspace, repo)}{_wrap_uuid(uuid)}")


def _pipeline_steps_by_uuid(
    client: BBClient, workspace: str, repo: str, pipeline_uuid: str
) -> list[dict[str, Any]]:
    """Internal: list steps when you already have a pipeline UUID. Used by
    `_resolve_step_uuid` (which already paid the build_number→uuid lookup)
    to avoid a second list-pipelines walk."""
    uuid = _strip_uuid_braces(pipeline_uuid)
    path = f"{_pipelines_root(workspace, repo)}{_wrap_uuid(uuid)}/steps/"
    return list(client.paginate(path, query={"pagelen": _BITBUCKET_MAX_PAGELEN}))


def pipeline_steps(
    client: BBClient, workspace: str, repo: str, build_number: int
) -> list[dict[str, Any]]:
    """List the steps of a pipeline by build_number."""
    uuid = _resolve_pipeline_uuid(client, workspace, repo, build_number)
    return _pipeline_steps_by_uuid(client, workspace, repo, uuid)


def pipeline_trigger(
    client: BBClient,
    workspace: str,
    repo: str,
    *,
    branch: str,
    pattern: str | None = None,
    variables: dict[str, str] | Iterable[tuple[str, str]] | None = None,
) -> dict[str, Any]:
    """Trigger a new pipeline run.

    Without `pattern`, runs the branch's default pipeline.
    With `pattern`, runs the named custom pipeline (must be defined in
    bitbucket-pipelines.yml under `custom:`).

    `variables` is the set of pipeline variables to pass — a dict
    {name: value} or an iterable of (name, value) tuples. Values must
    be strings; Bitbucket does not accept other JSON types for variables.

    Returns the new pipeline's record (includes build_number, uuid, etc.).
    """
    if not branch or not isinstance(branch, str):
        raise ValueError(f"branch is required and must be a string, got {branch!r}")

    # `type: "pipeline_ref_target"` is REQUIRED on the target. Without it
    # Bitbucket can't classify the reference and 400s with "Unsupported
    # reference target provided 'pipeline_unknown_target'". This bites the
    # custom-pattern path hardest (verified live); the default-branch path
    # happened to be accepted without it, but the field is correct for
    # both, so send it unconditionally for one consistent shape.
    target: dict[str, Any] = {
        "type": "pipeline_ref_target",
        "ref_type": "branch",
        "ref_name": branch,
    }
    if pattern is not None:
        if not isinstance(pattern, str) or not pattern:
            raise ValueError(f"pattern must be a non-empty string, got {pattern!r}")
        target["selector"] = {"type": "custom", "pattern": pattern}

    payload: dict[str, Any] = {"target": target}

    if variables is not None:
        # Normalise to a list of {"key": k, "value": v} dicts — Bitbucket's
        # contract. Accept both dict and iterable-of-pairs at the Python
        # boundary so MCP tool args can use either.
        if isinstance(variables, dict):
            items = list(variables.items())
        else:
            items = list(variables)
        normalised: list[dict[str, str]] = []
        for k, v in items:
            if not isinstance(k, str) or not k:
                raise ValueError(f"variable key must be a non-empty string, got {k!r}")
            if not isinstance(v, str):
                raise ValueError(
                    f"variable value for {k!r} must be a string, got {type(v).__name__}"
                )
            normalised.append({"key": k, "value": v})
        if normalised:
            payload["variables"] = normalised

    return client.post(_pipelines_root(workspace, repo), json_body=payload)


def pipeline_stop(
    client: BBClient, workspace: str, repo: str, build_number: int
) -> Any:
    """Stop a running pipeline. Returns the raw API response (typically
    None on success — Bitbucket returns 204). The bash script discards
    this response with `> /dev/null`; we return it so the MCP tool can
    surface a structured outcome (parity follow-up for 4.7)."""
    uuid = _resolve_pipeline_uuid(client, workspace, repo, build_number)
    path = f"{_pipelines_root(workspace, repo)}{_wrap_uuid(uuid)}/stopPipeline"
    return client.post(path)


def pipeline_logs(
    client: BBClient,
    workspace: str,
    repo: str,
    build_number: int,
    step_index: int,
    *,
    timeout: float = 120.0,
) -> str:
    """Fetch raw log text for a pipeline step (0-based step index).

    Bitbucket returns either the log body inline (200) or a 307 redirect
    to an S3 signed URL. The fetch helper follows redirects while
    stripping the Authorization header on cross-host hops so the
    Bitbucket credential is never sent to S3. Default timeout is 120s
    because log payloads can be large and the bash equivalent uses
    no timeout cap.
    """
    pipeline_uuid = _resolve_pipeline_uuid(client, workspace, repo, build_number)
    step_uuid = _resolve_step_uuid(
        client, workspace, repo, pipeline_uuid, step_index
    )
    path = (
        f"{_pipelines_root(workspace, repo)}"
        f"{_wrap_uuid(pipeline_uuid)}/steps/{_wrap_uuid(step_uuid)}/log"
    )
    return client.fetch_redirected_text(path, timeout=timeout)


# --- Pipelines configuration (enable / disable / status) ---
#
# A repo's Pipelines feature is toggled via the pipelines_config resource:
#   GET /repositories/{ws}/{slug}/pipelines_config   → {"enabled": bool, ...}
#   PUT  ...                          {"enabled": true|false} → updated config
# Pipelines must be ENABLED before repo pipeline variables, custom
# pipelines, or builds work at all — this is the "CI won't run / vars
# won't take" gap. Note (verified live): the GET 404s when Pipelines has
# never been configured on the repo (the pre-enable state), rather than
# returning {"enabled": false}. pipelines_config_show translates that 404
# into a clean {"enabled": false, "configured": false} so callers get a
# definite answer instead of an exception for the common "never enabled"
# case.


def pipelines_config_show(
    client: BBClient, workspace: str, repo: str
) -> dict[str, Any]:
    """Return the repo's Pipelines configuration: `{"enabled": bool, ...}`.

    GET /repositories/{ws}/{slug}/pipelines_config. When Pipelines has
    never been configured the API 404s; this is the normal "never enabled"
    state, so it's translated to `{"enabled": False, "configured": False}`
    rather than propagating BBApiError. When the config exists, the raw
    record is returned with a `"configured": True` marker added.
    """
    path = f"{repo_path(workspace, repo)}/pipelines_config"
    try:
        result = client.get(path)
    except BBApiError as e:
        if e.status == 404:
            return {"enabled": False, "configured": False}
        raise
    if isinstance(result, dict):
        result.setdefault("configured", True)
    return result


def pipelines_config_set(
    client: BBClient, workspace: str, repo: str, *, enabled: bool
) -> dict[str, Any]:
    """Enable or disable Pipelines on a repo.

    PUT /repositories/{ws}/{slug}/pipelines_config with
    `{"enabled": true|false}`. Returns the updated configuration record.

    Requires `admin:pipeline:bitbucket` scope on the token (toggling the
    Pipelines feature is a pipeline-admin operation, same scope family as
    `vars_set`). `write:pipeline:bitbucket` alone is insufficient; a 403
    names the missing scope under `error.detail.required`.
    """
    if not isinstance(enabled, bool):
        raise ValueError(f"enabled must be a bool, got {type(enabled).__name__}")
    path = f"{repo_path(workspace, repo)}/pipelines_config"
    return client.put(path, json_body={"enabled": enabled})


# ===========================================================================
#  PULL REQUEST OPERATIONS
# ===========================================================================

# Bitbucket's documented merge strategies. Validating at the boundary
# means the MCP tool fails fast on a typo rather than waiting for the
# server's 400.
_VALID_MERGE_STRATEGIES = frozenset({"merge_commit", "squash", "fast_forward"})

# PR `state` filter values the Bitbucket API accepts on the simple
# `?state=` query parameter. For multi-state filtering, Bitbucket requires
# the BBQL `q` parameter (e.g. `?q=state="OPEN" OR state="MERGED"`); the
# `?state=OPEN,MERGED` shape returns 400 / empty results. We validate the
# scalar form against this set when prs_list is called with `state=`;
# callers needing compound filtering should construct a `q=` query and
# call `client.paginate` directly.
_KNOWN_PR_STATES = frozenset({"OPEN", "MERGED", "DECLINED", "SUPERSEDED"})


def _prs_root(workspace: str, repo: str) -> str:
    """Common URL prefix for the pull-requests API."""
    return f"{repo_path(workspace, repo)}/pullrequests"


def _validate_pr_id(pr_id: int) -> None:
    """PR IDs are positive integers. The bash script passes them as bare
    strings and lets Bitbucket reject malformed values; we fail at the
    boundary so the MCP tool surfaces a clear error before any network
    call burns API budget.

    Rejects bool explicitly (`True`/`False` are subclass-of-int in Python
    but stringify to `"True"`/`"False"` in URLs, not `"1"`/`"0"`).
    """
    if not _is_positive_int(pr_id):
        raise ValueError(f"pr_id must be a positive int, got {pr_id!r}")


# Fields stripped from each PR object in the LIST view (prs_list) by
# default. Bitbucket PR objects carry the full rendered description +
# summary (raw / html / markup variants) and the participants array,
# which together push even a 3-PR list past the MCP 25k-token response
# cap on repos with rich PR bodies (observed: johnny-server, 3 open PRs
# = ~70 KB). The list/triage workflow (prs_list -> pick one -> pr_show)
# only needs identity + state + branches + author + links; the full
# body is one pr_show away. pr_show is intentionally NOT slimmed — it's
# the drill-down where you WANT the whole object.
_PR_LIST_BULKY_FIELDS = ("description", "summary", "rendered", "participants")


def _slim_pr_list_item(pr: dict[str, Any]) -> dict[str, Any]:
    """Drop the bulky fields from one PR list object. Shallow copy so
    the caller's source dict is untouched. `reviewers`, when present,
    is projected down to uuid + display_name per reviewer (the full
    account blobs are the other big contributor) while preserving the
    count and identities a triage view needs."""
    slim = {k: v for k, v in pr.items() if k not in _PR_LIST_BULKY_FIELDS}
    reviewers = pr.get("reviewers")
    if isinstance(reviewers, list):
        slim["reviewers"] = [
            {
                "uuid": r.get("uuid"),
                "display_name": r.get("display_name"),
            }
            for r in reviewers
            if isinstance(r, dict)
        ]
    return slim


def prs_list(
    client: BBClient,
    workspace: str,
    repo: str,
    *,
    state: str = "OPEN",
    count: int = 25,
    verbose: bool = False,
) -> list[dict[str, Any]]:
    """List pull requests filtered by state. Defaults match bash:
    state=OPEN, count=25. Walks pages as needed to honour `count`.

    By default each PR is slimmed (see _slim_pr_list_item) so the list
    fits the MCP response cap on rich-PR repos. Pass verbose=True to get
    the full Bitbucket PR objects (description, summary, rendered,
    participants intact) — useful when a caller genuinely needs the
    bodies and isn't going through the MCP transport."""
    if not _is_positive_int(count):
        raise ValueError(f"count must be a positive int, got {count!r}")
    if not isinstance(state, str) or not state:
        raise ValueError(f"state must be a non-empty string, got {state!r}")
    # _KNOWN_PR_STATES is the boundary check. Without it, typos like
    # state="OPENED", case bugs like state="open", and unsupported
    # compound forms like state="OPEN,MERGED" would burn an API call
    # before failing (Bitbucket returns 400 or empty results). For
    # compound filtering use a `?q=` query via client.paginate directly.
    if state not in _KNOWN_PR_STATES:
        raise ValueError(
            f"state must be one of {sorted(_KNOWN_PR_STATES)}, got {state!r}"
        )

    pagelen = min(count, _BITBUCKET_MAX_PAGELEN)
    query: dict[str, Any] = {"state": state, "pagelen": pagelen}

    out: list[dict[str, Any]] = []
    for pr in client.paginate(_prs_root(workspace, repo), query=query):
        out.append(pr if verbose else _slim_pr_list_item(pr))
        if len(out) >= count:
            break
    return out


def pr_show(
    client: BBClient, workspace: str, repo: str, pr_id: int
) -> dict[str, Any]:
    """Fetch a pull request by its numeric ID."""
    _validate_pr_id(pr_id)
    return client.get(f"{_prs_root(workspace, repo)}/{pr_id}")


def pr_activity(
    client: BBClient,
    workspace: str,
    repo: str,
    pr_id: int,
    *,
    count: int = 50,
) -> list[dict[str, Any]]:
    """List the activity stream on a PR (approvals, comment events, state
    transitions). Used by the bash `bb pr` to surface approver names;
    surfaced separately as an op so the MCP agent can render its own view
    of the activity timeline."""
    _validate_pr_id(pr_id)
    if not _is_positive_int(count):
        raise ValueError(f"count must be a positive int, got {count!r}")
    pagelen = min(count, _BITBUCKET_MAX_PAGELEN)
    out: list[dict[str, Any]] = []
    for entry in client.paginate(
        f"{_prs_root(workspace, repo)}/{pr_id}/activity",
        query={"pagelen": pagelen},
    ):
        out.append(entry)
        if len(out) >= count:
            break
    return out


def pr_create(
    client: BBClient,
    workspace: str,
    repo: str,
    *,
    title: str,
    source_branch: str,
    destination_branch: str = "main",
    description: str = "",
    close_source_branch: bool = True,
    reviewers: Iterable[str] | None = None,
) -> dict[str, Any]:
    """Create a pull request.

    `reviewers` is an iterable of Bitbucket account UUIDs (the API expects
    `[{"uuid": "..."}, ...]`). The bash script doesn't expose reviewers
    at create-time — that's a 4.7 parity gap, not a Python bug.

    `close_source_branch=True` matches bash's default (it hardcodes that
    flag in the create payload). If you don't want the branch deleted on
    merge, pass False explicitly.
    """
    for label, value in (
        ("title", title),
        ("source_branch", source_branch),
        ("destination_branch", destination_branch),
    ):
        # Strip-check rather than truthiness so " " / "\n\t" don't slip
        # through. A whitespace-only PR title is technically accepted by
        # Bitbucket but visually meaningless in any PR list view.
        if not isinstance(value, str) or not value.strip():
            raise ValueError(
                f"{label} must be a non-empty, non-whitespace string, got {value!r}"
            )
    if not isinstance(description, str):
        raise ValueError(
            f"description must be a string, got {type(description).__name__}"
        )
    if not isinstance(close_source_branch, bool):
        raise ValueError(
            f"close_source_branch must be a bool, got {type(close_source_branch).__name__}"
        )

    payload: dict[str, Any] = {
        "title": title,
        "source": {"branch": {"name": source_branch}},
        "destination": {"branch": {"name": destination_branch}},
        "close_source_branch": close_source_branch,
    }
    # Bash includes an empty description string ALWAYS; Python omits
    # when the description is empty or whitespace-only so the API payload
    # stays meaningful. Parity item: bash should align on omission.
    if description.strip():
        payload["description"] = description
    if reviewers is not None:
        # A bare string is technically an Iterable[str] (yields characters),
        # which would silently produce `[{"uuid":"a"}, {"uuid":"l"}, ...]`
        # from `reviewers="alice-uuid"`. Reject explicitly so the typo
        # fails locally rather than as a 400 from Bitbucket.
        if isinstance(reviewers, str):
            raise ValueError(
                f"reviewers must be a list/tuple of uuids, not a bare string. "
                f"Got {reviewers!r}; did you mean [{reviewers!r}]?"
            )
        normalised: list[dict[str, str]] = []
        for uuid in reviewers:
            if not isinstance(uuid, str) or not uuid:
                raise ValueError(
                    f"reviewer uuids must be non-empty strings, got {uuid!r}"
                )
            normalised.append({"uuid": uuid})
        if normalised:
            payload["reviewers"] = normalised

    return client.post(_prs_root(workspace, repo), json_body=payload)


def pr_update(
    client: BBClient,
    workspace: str,
    repo: str,
    pr_id: int,
    *,
    title: str | None = None,
    description: str | None = None,
) -> dict[str, Any]:
    """Update an OPEN pull request's title and/or description.

    Bitbucket's PR mutation endpoint is a PUT to the SAME path pr_show GETs:
    PUT /repositories/{ws}/{slug}/pullrequests/{id}. Only open pull requests
    can be mutated. Only the fields present in the body change — the PUT
    merges them into the existing PR, preserving the source/destination
    branches and reviewers (Bitbucket keeps omitted PR fields on this
    endpoint, so a title-only or description-only update is safe).

    Note the HTTP method is PUT, not PATCH: Bitbucket Cloud has no PATCH for
    the pullrequests resource. This mirrors repo_update, which PUTs the same
    path repo_show GETs for the analogous repo mutation.

    `title`: None (omit) = leave unchanged; any non-empty, non-whitespace
    string = set it. An empty/whitespace title is INVALID (a PR must have a
    title — Bitbucket rejects a blank one and it's meaningless in any PR
    list view), so it's rejected at the boundary rather than sent.

    `description`: None (omit) = leave unchanged; "" = intentionally CLEAR
    the body; any other string = set it. This three-way distinction matches
    repo_update's description contract so a deliberate clear isn't collapsed
    into a no-op.

    At least one of `title` / `description` must be supplied; a PUT with an
    empty body would be a no-op round-trip, so it's rejected at the boundary
    before burning an API call.

    Returns the updated PR record.
    """
    _validate_pr_id(pr_id)

    payload: dict[str, Any] = {}
    if title is not None:
        if not isinstance(title, str) or not title.strip():
            raise ValueError(
                f"title must be a non-empty, non-whitespace string when "
                f"provided, got {title!r}"
            )
        payload["title"] = title
    if description is not None:
        if not isinstance(description, str):
            raise ValueError(
                f"description must be a string when provided, got "
                f"{type(description).__name__}"
            )
        payload["description"] = description

    if not payload:
        raise ValueError(
            "pr_update requires at least one field to change "
            "(title and/or description)"
        )

    return client.put(f"{_prs_root(workspace, repo)}/{pr_id}", json_body=payload)


def pr_approve(
    client: BBClient, workspace: str, repo: str, pr_id: int
) -> Any:
    """Approve a pull request as the authenticated user. Returns the
    approval record; the bash equivalent discards it with `> /dev/null`."""
    _validate_pr_id(pr_id)
    return client.post(f"{_prs_root(workspace, repo)}/{pr_id}/approve")


def pr_unapprove(
    client: BBClient, workspace: str, repo: str, pr_id: int
) -> Any:
    """Remove the authenticated user's approval from a PR.

    Not exposed by the bash CLI today — this is one of the parity gaps
    that 4.7 will fill. The Bitbucket REST contract is a DELETE against
    the same /approve subpath that POST uses for approval.
    """
    _validate_pr_id(pr_id)
    return client.delete(f"{_prs_root(workspace, repo)}/{pr_id}/approve")


def pr_merge(
    client: BBClient,
    workspace: str,
    repo: str,
    pr_id: int,
    *,
    strategy: str = "merge_commit",
    close_source_branch: bool = True,
    message: str | None = None,
) -> dict[str, Any]:
    """Merge a pull request.

    Bitbucket Cloud's documented strategies: `merge_commit` (default),
    `squash`, `fast_forward`. We validate at the boundary so a typo
    fails locally rather than burning an API call to get a 400.

    `message` overrides the default merge-commit message. `close_source_branch`
    matches bash's default of True.
    """
    _validate_pr_id(pr_id)
    # isinstance gate before the membership test: a non-hashable strategy
    # (list, dict, set) would otherwise raise TypeError from the frozenset
    # `in` check rather than the documented ValueError, breaking the
    # "every boundary failure is ValueError" convention this file follows.
    if not isinstance(strategy, str) or strategy not in _VALID_MERGE_STRATEGIES:
        raise ValueError(
            f"strategy must be one of {sorted(_VALID_MERGE_STRATEGIES)}, "
            f"got {strategy!r}"
        )
    if not isinstance(close_source_branch, bool):
        raise ValueError(
            f"close_source_branch must be a bool, got {type(close_source_branch).__name__}"
        )
    payload: dict[str, Any] = {
        "type": "pullrequest",
        "merge_strategy": strategy,
        "close_source_branch": close_source_branch,
    }
    if message is not None:
        # Symmetric with pr_comment_add's body validation: empty (or
        # whitespace-only) message would produce a blank merge-commit
        # subject line, visually empty in any `git log --oneline` view.
        # Reject at the boundary.
        if not isinstance(message, str) or not message.strip():
            raise ValueError(
                f"message must be a non-empty, non-whitespace string "
                f"when provided, got {message!r}"
            )
        payload["message"] = message
    # Bitbucket's PR merge endpoint is POST per the REST docs. An earlier
    # version of this op (and bash cmd_pr_merge) used PUT on the
    # historical assumption that Bitbucket accepted either; that's no
    # longer true. PUT now returns HTTP 403 + "This endpoint does not
    # support token-based authentication" (an unhelpful error that
    # actually means "wrong method here"). Confirmed against
    # dreamfacesbir/ryan-os PR#2 (2026-06-28) where direct POST with the
    # same API token merged cleanly. Keep POST; do not "improve" back.
    return client.post(
        f"{_prs_root(workspace, repo)}/{pr_id}/merge",
        json_body=payload,
    )


def pr_decline(
    client: BBClient, workspace: str, repo: str, pr_id: int
) -> Any:
    """Decline (close without merging) a pull request."""
    _validate_pr_id(pr_id)
    return client.post(f"{_prs_root(workspace, repo)}/{pr_id}/decline")


def pr_diff(
    client: BBClient,
    workspace: str,
    repo: str,
    pr_id: int,
    *,
    timeout: float = 120.0,
) -> str:
    """Fetch the unified diff text for a pull request.

    Bitbucket returns plain text (not JSON), so we route through
    `fetch_redirected_text`. Today the diff endpoint does NOT redirect,
    so this is functionally equivalent to a direct GET. If Bitbucket
    ever introduces a redirect, the cross-host-auth-strip protection
    kicks in — but the returned body would then be whatever the redirect
    target serves (a behavioural divergence from bash, which uses
    `curl -sf` without `-L` and would fail visibly on any 3xx). Until
    that happens, the two surfaces produce identical text.
    """
    _validate_pr_id(pr_id)
    return client.fetch_redirected_text(
        f"{_prs_root(workspace, repo)}/{pr_id}/diff",
        timeout=timeout,
    )


def pr_comments_list(
    client: BBClient,
    workspace: str,
    repo: str,
    pr_id: int,
    *,
    count: int = 100,
) -> list[dict[str, Any]]:
    """List comments on a pull request."""
    _validate_pr_id(pr_id)
    if not _is_positive_int(count):
        raise ValueError(f"count must be a positive int, got {count!r}")
    pagelen = min(count, _BITBUCKET_MAX_PAGELEN)
    out: list[dict[str, Any]] = []
    for comment in client.paginate(
        f"{_prs_root(workspace, repo)}/{pr_id}/comments",
        query={"pagelen": pagelen},
    ):
        out.append(comment)
        if len(out) >= count:
            break
    return out


def pr_comment_add(
    client: BBClient,
    workspace: str,
    repo: str,
    pr_id: int,
    body: str,
) -> dict[str, Any]:
    """Add a top-level comment to a pull request.

    Not exposed by the bash CLI today — 4.7 parity gap. The Bitbucket
    contract is `POST /pullrequests/{id}/comments` with payload
    `{"content": {"raw": "<text>"}}`.
    """
    _validate_pr_id(pr_id)
    if not isinstance(body, str) or not body.strip():
        raise ValueError(
            f"body must be a non-empty, non-whitespace string, got {body!r}"
        )
    return client.post(
        f"{_prs_root(workspace, repo)}/{pr_id}/comments",
        json_body={"content": {"raw": body}},
    )


# ===========================================================================
#  WORKSPACES
# ===========================================================================


def workspaces_list(
    client: BBClient,
    *,
    count: int = 100,
) -> list[dict[str, Any]]:
    """List the Bitbucket workspaces the authenticated user belongs to.

    Uses `GET /2.0/user/workspaces` — the CHANGE-3022 replacement for
    the cross-workspace listing endpoints removed under CHANGE-2770
    (effective 2026-04-14). The old `/2.0/workspaces` and
    `/2.0/user/permissions/workspaces` both now return CHANGE-2770
    errors regardless of token shape.

    Requires `read:workspace:bitbucket` scope on the API token. A token
    granted only repository/pullrequest/pipeline scopes will surface
    Bitbucket's "credentials lack one or more required privilege
    scopes" 403 verbatim through the BBApiError path — the agent /
    user sees exactly which scope to add when rotating.

    Each value is a `workspace_access` envelope with the new sparse
    schema: `.administrator` (bool), `.workspace.slug`, `.workspace.uuid`,
    `.workspace.links` (no `name` / no `permission` string — those were
    legacy fields not carried into the new endpoint). Callers should
    branch on `administrator` (bool) rather than expecting a role
    string.
    """
    if not _is_positive_int(count):
        raise ValueError(f"count must be a positive int, got {count!r}")

    pagelen = min(count, _BITBUCKET_MAX_PAGELEN)
    q: dict[str, Any] = {"pagelen": pagelen}

    out: list[dict[str, Any]] = []
    for w in client.paginate("/user/workspaces", query=q):
        out.append(w)
        if len(out) >= count:
            break
    return out


def projects_list(
    client: BBClient,
    workspace: str | None = None,
    *,
    count: int = 100,
) -> list[dict[str, Any]]:
    """List the projects in a workspace.

    Uses `GET /2.0/workspaces/{workspace}/projects` (paginated). Each
    value is a project record carrying `.key` (the short project key used
    in repo bodies, e.g. `WID`), `.name`, `.uuid`, and `.links`.

    `workspace=None` defaults to the client's configured workspace
    (`client.config.workspace`); pass an explicit workspace to query a
    different one.

    Requires the `read:project:bitbucket` scope on the API token. A token
    without it surfaces Bitbucket's "credentials lack one or more required
    privilege scopes" 403 verbatim through the BBApiError path; the exact
    missing scope is recoverable from the error body under
    `error.detail.required`.
    """
    ws = workspace if workspace is not None else client.config.workspace
    # Mirror repos_list: this op doesn't route through repo_path (it's a
    # workspace-level listing, no slug), so the workspace contract is
    # validated inline — empty/whitespace AND embedded `/`, `.`, `..`.
    if not isinstance(ws, str) or not ws.strip():
        raise ValueError(f"workspace must be a non-empty string, got {ws!r}")
    if "/" in ws:
        raise ValueError(f"workspace must not contain '/', got {ws!r}")
    if ws in (".", ".."):
        raise ValueError(f"workspace must not be '.' or '..', got {ws!r}")
    if not _is_positive_int(count):
        raise ValueError(f"count must be a positive int, got {count!r}")

    pagelen = min(count, _BITBUCKET_MAX_PAGELEN)
    q: dict[str, Any] = {"pagelen": pagelen}

    out: list[dict[str, Any]] = []
    for p in client.paginate(f"/workspaces/{ws}/projects", query=q):
        out.append(p)
        if len(out) >= count:
            break
    return out


# ===========================================================================
#  REPOSITORY / BRANCH / VARIABLES / DOWNLOADS / COMMITS
# ===========================================================================


def repos_list(
    client: BBClient,
    workspace: str | None = None,
    *,
    count: int = 100,
    sort: str = "-updated_on",
    query: str | None = None,
) -> list[dict[str, Any]]:
    """List repositories in a workspace.

    `workspace=None` defaults to the client's configured workspace
    (`client.config.workspace`); pass an explicit workspace to query
    a different one (the bash equivalent only ever uses BB_WORKSPACE).

    `query` is a Bitbucket BBQL filter string passed via `?q=`, e.g.
    `'name ~ "widget"'`. Bash doesn't expose this; it's a 4.7 parity
    gap for the agent's filtered-list workflows.
    """
    ws = workspace if workspace is not None else client.config.workspace
    # Symmetric with bb_api.repo_path: reject empty/whitespace AND
    # embedded `/`, `.`, `..`. Without this, `workspace="acme/widget"`
    # would silently build `/repositories/acme/widget` (a single-repo
    # endpoint), then paginate against a response that lacks `values`
    # — a confusing failure mode the boundary validator exists to
    # prevent everywhere else in this file. repos_list is the only op
    # that doesn't route through repo_path, so the check is duplicated
    # here rather than central.
    if not isinstance(ws, str) or not ws.strip():
        raise ValueError(f"workspace must be a non-empty string, got {ws!r}")
    if "/" in ws:
        raise ValueError(f"workspace must not contain '/', got {ws!r}")
    if ws in (".", ".."):
        raise ValueError(f"workspace must not be '.' or '..', got {ws!r}")
    if not _is_positive_int(count):
        raise ValueError(f"count must be a positive int, got {count!r}")

    pagelen = min(count, _BITBUCKET_MAX_PAGELEN)
    q: dict[str, Any] = {"sort": sort, "pagelen": pagelen}
    if query is not None:
        if not isinstance(query, str) or not query.strip():
            raise ValueError(
                f"query must be a non-empty, non-whitespace string when provided, "
                f"got {query!r}"
            )
        q["q"] = query

    out: list[dict[str, Any]] = []
    for r in client.paginate(f"/repositories/{ws}", query=q):
        out.append(r)
        if len(out) >= count:
            break
    return out


def repo_show(
    client: BBClient, workspace: str, repo: str
) -> dict[str, Any]:
    """Fetch repository metadata: language, size, clone URLs, mainbranch,
    privacy, etc."""
    return client.get(repo_path(workspace, repo))


def repo_create(
    client: BBClient,
    workspace: str,
    repo: str,
    *,
    is_private: bool = True,
    project_key: str | None = None,
    description: str | None = None,
) -> dict[str, Any]:
    """Create a new repository via POST /repositories/{workspace}/{repo}.

    Bitbucket's create endpoint is a POST to the SAME path repo_show GETs;
    the repo slug is in the URL and the body carries the settings. The
    `scm` field is always "git" (Bitbucket Cloud dropped Mercurial), and
    `is_private` defaults to true so a forgotten flag never publishes a
    repo by accident.

    `project_key` maps to the body's `{"project": {"key": ...}}`. A
    workspace that has any projects REQUIRES one; Bitbucket 400s a
    create with no project on such a workspace, so the caller surfaces
    the error rather than this function inventing a default.

    Returns the created repo record (includes slug, full_name, links —
    the clone URLs are under `.links.clone`).
    """
    # repo_path validates workspace + slug at the boundary (empty,
    # whitespace, embedded '/', '.', '..') — reuse it so create enforces
    # the exact same slug contract as every other repo op.
    path = repo_path(workspace, repo)

    payload: dict[str, Any] = {"scm": "git", "is_private": bool(is_private)}
    if project_key is not None:
        if not isinstance(project_key, str) or not project_key.strip():
            raise ValueError(
                f"project_key must be a non-empty, non-whitespace string when "
                f"provided, got {project_key!r}"
            )
        payload["project"] = {"key": project_key.strip()}
    if description is not None:
        if not isinstance(description, str):
            raise ValueError(
                f"description must be a string when provided, got "
                f"{type(description).__name__}"
            )
        payload["description"] = description

    return client.post(path, json_body=payload)


def repo_update(
    client: BBClient,
    workspace: str,
    repo: str,
    *,
    project_key: str | None = None,
    description: str | None = None,
) -> dict[str, Any]:
    """Update an existing repository via PUT /repositories/{workspace}/{repo}.

    Bitbucket's update endpoint is a PUT to the SAME path repo_show GETs
    and repo_create POSTs; only the fields present in the body change. The
    dominant use is moving a repo between projects (repo_create takes a
    project but nothing could change it afterward — this closes that gap).

    `project_key` maps to the body's `{"project": {"key": ...}}` — that's
    how Bitbucket reassigns a repo's project. `description` updates the
    repo description. At least one field must be supplied; a PUT with an
    empty body would be a no-op round-trip, so it's rejected at the
    boundary rather than burning an API call.

    Requires `admin:repository:bitbucket` scope on the token (same as
    repo_create — changing repo settings is an admin operation).
    `write:repository:bitbucket` alone returns 403, whose body names the
    missing scope under `error.detail.required`.

    Returns the updated repo record.
    """
    # repo_path validates workspace + slug at the boundary (empty,
    # whitespace, embedded '/', '.', '..') — reuse it so update enforces
    # the exact same slug contract as every other repo op.
    path = repo_path(workspace, repo)

    payload: dict[str, Any] = {}
    if project_key is not None:
        if not isinstance(project_key, str) or not project_key.strip():
            raise ValueError(
                f"project_key must be a non-empty, non-whitespace string when "
                f"provided, got {project_key!r}"
            )
        payload["project"] = {"key": project_key.strip()}
    if description is not None:
        if not isinstance(description, str):
            raise ValueError(
                f"description must be a string when provided, got "
                f"{type(description).__name__}"
            )
        payload["description"] = description

    if not payload:
        raise ValueError(
            "repo_update requires at least one field to change "
            "(project_key and/or description)"
        )

    return client.put(path, json_body=payload)


# --- Deployment environments ---
#
# Deployment environments are the named targets a pipeline deploys to
# (Test / Staging / Production), each carrying its own deployment
# variables (covered separately by vars_set --deployment). These ops
# manage the environments themselves:
#   GET    /repositories/{ws}/{slug}/environments/            → list
#   POST   ...   {"name", "environment_type": {"name"}}       → create
#   DELETE ...   /{env_uuid}/                                 → delete
# Body shape + 201/204 responses verified against the live API. The
# environment_type.name is one of Bitbucket's deployment categories:
# Test, Staging, Production.

# Bitbucket's deployment environment categories. The POST body's
# environment_type.name must be one of these (verified live: "Test"
# accepted). Compared case-insensitively, sent in Bitbucket's canonical
# capitalisation.
_ENVIRONMENT_TYPES = {"test": "Test", "staging": "Staging", "production": "Production"}


def environments_list(
    client: BBClient, workspace: str, repo: str, *, count: int = 100
) -> list[dict[str, Any]]:
    """List the repo's deployment environments.

    GET /repositories/{ws}/{slug}/environments/ (paginated). Each record
    carries `.name`, `.slug`, `.uuid`, and `.environment_type`.
    """
    if not _is_positive_int(count):
        raise ValueError(f"count must be a positive int, got {count!r}")
    pagelen = min(count, _BITBUCKET_MAX_PAGELEN)
    out: list[dict[str, Any]] = []
    for env in client.paginate(
        f"{repo_path(workspace, repo)}/environments/",
        query={"pagelen": pagelen},
    ):
        out.append(env)
        if len(out) >= count:
            break
    return out


def environment_create(
    client: BBClient,
    workspace: str,
    repo: str,
    name: str,
    *,
    environment_type: str = "Test",
) -> dict[str, Any]:
    """Create a deployment environment.

    POST /repositories/{ws}/{slug}/environments/ with
    `{"name": ..., "environment_type": {"name": ...}}`. `environment_type`
    is one of Test / Staging / Production (case-insensitive; sent in
    canonical capitalisation). Returns the created environment record
    (includes its `uuid`, needed for delete and for vars --deployment).

    Requires `admin:pipeline:bitbucket` scope on the token (deployment
    environments are part of the Pipelines admin surface). A 403 names the
    missing scope under `error.detail.required`.
    """
    if not isinstance(name, str) or not name.strip():
        raise ValueError(
            f"name must be a non-empty, non-whitespace string, got {name!r}"
        )
    if not isinstance(environment_type, str) or not environment_type.strip():
        raise ValueError(
            f"environment_type must be a non-empty string, got {environment_type!r}"
        )
    canonical = _ENVIRONMENT_TYPES.get(environment_type.strip().casefold())
    if canonical is None:
        raise ValueError(
            f"environment_type must be one of Test / Staging / Production, "
            f"got {environment_type!r}"
        )
    path = f"{repo_path(workspace, repo)}/environments/"
    payload = {"name": name.strip(), "environment_type": {"name": canonical}}
    return client.post(path, json_body=payload)


def environment_delete(
    client: BBClient, workspace: str, repo: str, environment: str
) -> Any:
    """Delete a deployment environment by NAME (or slug).

    Resolves the name to its UUID (reusing the same lookup as
    vars --deployment), then DELETEs
    /repositories/{ws}/{slug}/environments/{env_uuid}/. Returns the raw
    response (None on success — Bitbucket returns 204). Raises
    BBOpNotFound if no environment matches the name, so a typo fails
    clearly instead of silently no-op'ing.

    Requires `admin:pipeline:bitbucket` scope on the token (same as
    create). A 403 names the missing scope under `error.detail.required`.
    """
    uuid = _resolve_environment_uuid(client, workspace, repo, environment)
    encoded = quote(uuid, safe="")
    path = f"{repo_path(workspace, repo)}/environments/{encoded}/"
    return client.delete(path)


# --- Branches ---


def branches_list(
    client: BBClient,
    workspace: str,
    repo: str,
    *,
    count: int = 50,
    sort: str = "-target.date",
    query: str | None = None,
) -> list[dict[str, Any]]:
    """List branches in the repo, default sort is most-recently-updated
    first (matches bash). `query` is a Bitbucket BBQL filter for
    name-substring etc.; not exposed by bash."""
    if not _is_positive_int(count):
        raise ValueError(f"count must be a positive int, got {count!r}")
    pagelen = min(count, _BITBUCKET_MAX_PAGELEN)
    q: dict[str, Any] = {"sort": sort, "pagelen": pagelen}
    if query is not None:
        if not isinstance(query, str) or not query.strip():
            raise ValueError(
                f"query must be a non-empty, non-whitespace string when provided, "
                f"got {query!r}"
            )
        q["q"] = query

    out: list[dict[str, Any]] = []
    for br in client.paginate(
        f"{repo_path(workspace, repo)}/refs/branches", query=q
    ):
        out.append(br)
        if len(out) >= count:
            break
    return out


def branch_show(
    client: BBClient, workspace: str, repo: str, name: str
) -> dict[str, Any]:
    """Fetch a single branch by name. Not exposed by bash today — 4.7
    parity gap. Useful for the agent's "does this branch exist?" lookup
    before creating a PR.

    The branch name is URL-encoded; `feat/widget` becomes `feat%2Fwidget`
    in the request path so the slash isn't interpreted as a sub-resource.
    """
    if not isinstance(name, str) or not name.strip():
        raise ValueError(
            f"name must be a non-empty, non-whitespace string, got {name!r}"
        )
    # `quote(s, safe="")` URL-encodes `/` to `%2F`. Branch names like
    # `feat/widget` would otherwise be interpreted as a sub-resource
    # path by Bitbucket and 404.
    encoded = quote(name.strip(), safe="")
    return client.get(f"{repo_path(workspace, repo)}/refs/branches/{encoded}")


# --- Variables (pipeline configuration) ---
#
# Bitbucket exposes pipeline variables at THREE scopes, each on a
# different (and inconsistently-named) endpoint. The base paths are
# verified against the live API:
#
#   repo        /repositories/{ws}/{slug}/pipelines_config/variables/
#   workspace   /workspaces/{ws}/pipelines-config/variables/      (HYPHEN)
#   deployment  /repositories/{ws}/{slug}/deployments_config/
#                   environments/{env_uuid}/variables/
#
# Note the workspace scope uses `pipelines-config` (hyphen) where the
# repo scope uses `pipelines_config` (underscore); the underscore form
# 404s at the workspace scope. The deployment scope keys off an
# environment UUID, which the caller supplies by NAME — _resolve_environment_uuid
# maps it. All three share the same body shape ({key, value, secured})
# and the same create-or-update semantics, so the list / find / set
# helpers take a pre-built `base` path and are scope-agnostic.

_VARS_SCOPES = ("repo", "workspace", "deployment")

# Sentinel for vars_set's `existing` param. `None` is a MEANINGFUL value
# there: it's what a caller passes after looking the key up and finding it
# ABSENT (so vars_set should go straight to create, no second lookup).
# This sentinel is the real "caller did not pre-fetch" marker, so vars_set
# only re-runs the lookup when nothing was passed. Without it, a caller
# that pre-fetched and got None would trigger a redundant second
# pagination on every create.
_NOT_PREFETCHED: Any = object()


def _resolve_environment_uuid(
    client: BBClient, workspace: str, repo: str, environment: str
) -> str:
    """Resolve a deployment-environment NAME (or slug) to its UUID.

    The deployment-variable endpoint is keyed by environment UUID, but
    humans refer to environments by name ("Development", "Production").
    GET the environment list and match case-insensitively on `name`
    first, then `slug`. Raise BBOpNotFound if no environment matches so
    the caller surfaces a clear "no such environment" instead of building
    a URL with an empty UUID.

    Returns the brace-wrapped UUID exactly as Bitbucket returns it (the
    `set`/`list` helpers URL-encode it for the path).
    """
    if not isinstance(environment, str) or not environment.strip():
        raise ValueError(
            f"environment must be a non-empty, non-whitespace string, "
            f"got {environment!r}"
        )
    target = environment.strip().casefold()
    base = f"{repo_path(workspace, repo)}/environments/"
    names: list[str] = []
    for env in client.paginate(base, query={"pagelen": _BITBUCKET_MAX_PAGELEN}):
        name = env.get("name")
        slug = env.get("slug")
        if name:
            names.append(name)
        if (isinstance(name, str) and name.casefold() == target) or (
            isinstance(slug, str) and slug.casefold() == target
        ):
            uuid = env.get("uuid")
            if not uuid:
                raise BBOpNotFound(
                    f"environment {environment!r} found but has no uuid"
                )
            return uuid
    raise BBOpNotFound(
        f"no deployment environment named {environment!r} in "
        f"{workspace}/{repo} (available: {names or 'none'})"
    )


def _variables_base(
    client: BBClient,
    workspace: str,
    repo: str | None,
    *,
    scope: str = "repo",
    environment: str | None = None,
) -> str:
    """Build the variables collection base path for the requested scope.

    `repo` is required for the repo and deployment scopes; ignored for
    the workspace scope. `environment` is required for the deployment
    scope (resolved to a UUID here) and rejected for the others.
    """
    if scope not in _VARS_SCOPES:
        raise ValueError(
            f"scope must be one of {_VARS_SCOPES}, got {scope!r}"
        )

    if scope == "workspace":
        if environment is not None:
            raise ValueError("environment is only valid for the deployment scope")
        # Validate the workspace the same way repo_path validates its
        # segments (empty / whitespace / '/' / '.' / '..').
        if not workspace or not workspace.strip():
            raise ValueError(
                f"workspace must be a non-empty, non-whitespace string, "
                f"got {workspace!r}"
            )
        if "/" in workspace:
            raise ValueError(f"workspace must not contain '/', got {workspace!r}")
        if workspace in (".", ".."):
            raise ValueError(f"workspace must not be '.' or '..', got {workspace!r}")
        # HYPHEN form — verified against the live API; the underscore
        # form 404s at the workspace scope.
        return f"/workspaces/{workspace}/pipelines-config/variables/"

    if repo is None:
        raise ValueError(f"repo is required for the {scope} scope")

    if scope == "repo":
        if environment is not None:
            raise ValueError("environment is only valid for the deployment scope")
        return f"{repo_path(workspace, repo)}/pipelines_config/variables/"

    # scope == "deployment"
    if environment is None:
        raise ValueError("environment is required for the deployment scope")
    env_uuid = _resolve_environment_uuid(client, workspace, repo, environment)
    encoded = quote(env_uuid, safe="")
    return (
        f"{repo_path(workspace, repo)}/deployments_config/"
        f"environments/{encoded}/variables/"
    )


def vars_list(
    client: BBClient,
    workspace: str,
    repo: str | None = None,
    *,
    count: int = 100,
    scope: str = "repo",
    environment: str | None = None,
) -> list[dict[str, Any]]:
    """List pipeline configuration variables (key/value pairs, with a
    `secured` flag that masks values for sensitive variables).

    `scope` selects repo (default), workspace, or deployment. The
    deployment scope requires `environment` (an env name/slug, resolved
    to a UUID). See `_variables_base` for the endpoint per scope.

    Bash truncates secured values to `********` in its display layer;
    Python returns the raw dicts (which include `"value": null` for
    secured entries — Bitbucket does NOT echo secured values). The
    MCP agent surfaces the secured flag explicitly so callers don't
    accidentally assume `null` means "unset".
    """
    if not _is_positive_int(count):
        raise ValueError(f"count must be a positive int, got {count!r}")
    base = _variables_base(
        client, workspace, repo, scope=scope, environment=environment
    )
    pagelen = min(count, _BITBUCKET_MAX_PAGELEN)
    out: list[dict[str, Any]] = []
    for v in client.paginate(base, query={"pagelen": pagelen}):
        out.append(v)
        if len(out) >= count:
            break
    return out


def _find_var_by_key_at(
    client: BBClient, base: str, key: str
) -> dict[str, Any] | None:
    """Return the existing variable dict at `base` whose `key` matches,
    or None. Walks the full list (no count cap) so a create-or-update
    never mistakes "not on page 1" for "doesn't exist" and POSTs a
    duplicate (Bitbucket allows duplicate keys via the API)."""
    for v in client.paginate(base, query={"pagelen": _BITBUCKET_MAX_PAGELEN}):
        if v.get("key") == key:
            return v
    return None


def vars_set(
    client: BBClient,
    workspace: str,
    repo: str | None,
    key: str,
    value: str,
    *,
    secured: bool = False,
    scope: str = "repo",
    environment: str | None = None,
    existing: dict[str, Any] | None = _NOT_PREFETCHED,
    base: str | None = None,
) -> dict[str, Any]:
    """Create or update a pipeline variable at the requested scope.

    Finds an existing variable by `key`. If present, PUTs to
    `.../variables/{uuid}` (update); otherwise POSTs to
    `.../variables/` (create). Body is `{"key", "value", "secured"}`.
    Works identically across the repo / workspace / deployment scopes —
    only the base path differs (see `_variables_base`).

    Secret hygiene: this function NEVER logs or echoes `value`. Bitbucket
    does not echo a secured variable's value back on write either (the
    response `value` is null when `secured` is true), so the returned
    dict is safe to surface — but callers must still mask the value they
    PASSED IN. The bash/MCP layers read the value from a file or env var
    so it never lands in argv, and they mask it in any output.

    Returns the created/updated variable record. For a secured variable
    Bitbucket returns `{"key", "uuid", "secured": true, "value": null}` —
    the absence of the value is by design, not an error.

    `existing` lets a caller that already looked the variable up (e.g.
    to report "created" vs "updated") pass the lookup result in so the
    full variable list isn't paginated twice. Pass the looked-up dict to
    update it, or `None` to signal "looked it up, it's absent" (go
    straight to create with no second pagination). Omit the argument
    entirely (the `_NOT_PREFETCHED` default) to have this function do the
    lookup. The distinction matters: `None` is "pre-fetched and absent",
    NOT "not pre-fetched"; conflating them re-paginated on every create.

    `base` lets a caller that already built the collection path (via
    `_variables_base`) pass it in so the deployment scope doesn't resolve
    the environment NAME->UUID twice (once in the caller's find-by-key,
    once here). When `None` the base is built from scope/environment.
    This keeps the deployment path to a single environment lookup,
    matching the bash CLI (which resolves the env once into `$base`).
    """
    if not isinstance(key, str) or not key.strip():
        raise ValueError(
            f"key must be a non-empty, non-whitespace string, got {key!r}"
        )
    if not isinstance(value, str):
        raise ValueError(
            f"value must be a string, got {type(value).__name__}"
        )
    key = key.strip()

    if base is None:
        base = _variables_base(
            client, workspace, repo, scope=scope, environment=environment
        )
    else:
        # Validate scope/environment consistency even when `base` is
        # supplied, so a caller passing a base that contradicts
        # scope/environment can't bypass the "environment required for the
        # deployment scope" / "environment only valid for deployment"
        # guards. These checks are pure (no network) for all scopes.
        if scope not in _VARS_SCOPES:
            raise ValueError(
                f"scope must be one of {_VARS_SCOPES}, got {scope!r}"
            )
        if scope == "deployment" and environment is None:
            raise ValueError("environment is required for the deployment scope")
        if scope != "deployment" and environment is not None:
            raise ValueError("environment is only valid for the deployment scope")

    payload: dict[str, Any] = {
        "key": key,
        "value": value,
        "secured": bool(secured),
    }

    # `_NOT_PREFETCHED` means the caller didn't look it up, so do it here.
    # A literal `None` means the caller looked it up and it's absent, so
    # skip straight to create (no redundant second pagination).
    if existing is _NOT_PREFETCHED:
        existing = _find_var_by_key_at(client, base, key)
    if existing is not None:
        uuid = existing.get("uuid")
        if not uuid:
            # An existing entry with no uuid can't be addressed for PUT.
            # Raise rather than fall through to POST (which would create
            # a duplicate-keyed variable).
            raise BBOpNotFound(
                f"variable {key!r} found but has no uuid; cannot update"
            )
        # uuid comes back from the API already brace-wrapped (`{...}`);
        # it must be URL-encoded for the path segment so the braces don't
        # break the URL.
        encoded_uuid = quote(uuid, safe="")
        return client.put(f"{base}{encoded_uuid}", json_body=payload)

    return client.post(base, json_body=payload)


def vars_delete(
    client: BBClient,
    workspace: str,
    repo: str | None,
    key: str,
    *,
    scope: str = "repo",
    environment: str | None = None,
) -> dict[str, Any]:
    """Delete a pipeline configuration variable by key.

    Resolves the variable by `key` to its UUID (walking all pages, same
    as `vars_set` — so "not on page 1" is never mistaken for "absent"),
    then DELETEs `.../variables/{uuid}`. Works identically across the
    repo / workspace / deployment scopes; only the base path differs (see
    `_variables_base`).

    Raises `BBOpNotFound` if no variable with that key exists at the scope,
    BEFORE issuing any DELETE — so a typo'd key is a clean not-found, not a
    silent no-op or a spurious network write.

    Requires `admin:pipeline:bitbucket` scope on the token (same as
    `vars_set` — they're the same pipeline-config resource);
    `write:pipeline:bitbucket` alone returns 403, whose body names the
    missing scope under `error.detail.required`.

    Returns `{"key", "scope", "environment", "uuid"}` so the caller can
    confirm exactly what was removed. The MCP/bash layers add
    `action: "deleted"`.
    """
    if not isinstance(key, str) or not key.strip():
        raise ValueError(
            f"key must be a non-empty, non-whitespace string, got {key!r}"
        )
    key = key.strip()

    base = _variables_base(
        client, workspace, repo, scope=scope, environment=environment
    )
    existing = _find_var_by_key_at(client, base, key)
    if existing is None:
        raise BBOpNotFound(
            f"no variable named {key!r} at the {scope} scope"
            + (f" (environment {environment!r})" if environment else "")
        )
    uuid = existing.get("uuid")
    if not uuid:
        # An entry with no uuid can't be addressed for DELETE — surface it
        # rather than building a `.../variables/` (collection) DELETE that
        # would 405 or, worse, do something unexpected.
        raise BBOpNotFound(
            f"variable {key!r} found but has no uuid; cannot delete"
        )
    encoded_uuid = quote(uuid, safe="")
    client.delete(f"{base}{encoded_uuid}")
    return {"key": key, "scope": scope, "environment": environment, "uuid": uuid}


# --- Downloads (release artifacts) ---


def downloads_list(
    client: BBClient,
    workspace: str,
    repo: str,
    *,
    count: int = 25,
) -> list[dict[str, Any]]:
    """List repository download artifacts (the Bitbucket "Downloads" tab
    — release binaries, install bundles, etc.)."""
    if not _is_positive_int(count):
        raise ValueError(f"count must be a positive int, got {count!r}")
    pagelen = min(count, _BITBUCKET_MAX_PAGELEN)
    out: list[dict[str, Any]] = []
    for d in client.paginate(
        f"{repo_path(workspace, repo)}/downloads",
        query={"pagelen": pagelen},
    ):
        out.append(d)
        if len(out) >= count:
            break
    return out


# --- Commits ---


def commits_list(
    client: BBClient,
    workspace: str,
    repo: str,
    *,
    branch: str | None = None,
    count: int = 10,
) -> list[dict[str, Any]]:
    """List recent commits.

    With `branch=None`, lists across all branches (Bitbucket's
    `/commits` endpoint).

    With `branch="feat/widget"`, lists commits reachable from that
    branch (`/commits/{branch}`). Branch names are URL-encoded for
    the same slash-as-sub-resource reason as branch_show.

    Not exposed by the bash CLI today — 4.7 parity gap. Useful for
    the agent's "what shipped recently?" / "what's in this branch
    that isn't in main?" workflows.
    """
    if not _is_positive_int(count):
        raise ValueError(f"count must be a positive int, got {count!r}")
    if branch is not None and (not isinstance(branch, str) or not branch.strip()):
        raise ValueError(
            f"branch must be a non-empty, non-whitespace string when provided, "
            f"got {branch!r}"
        )

    pagelen = min(count, _BITBUCKET_MAX_PAGELEN)
    if branch is None:
        path = f"{repo_path(workspace, repo)}/commits"
    else:
        encoded = quote(branch.strip(), safe="")
        path = f"{repo_path(workspace, repo)}/commits/{encoded}"

    out: list[dict[str, Any]] = []
    for c in client.paginate(path, query={"pagelen": pagelen}):
        out.append(c)
        if len(out) >= count:
            break
    return out