open-swe/agent/utils/authorship.py

"""Helpers for resolving the triggering user's git identity."""

from __future__ import annotations

import logging
from dataclasses import dataclass
from typing import Any

import httpx

logger = logging.getLogger(__name__)

# Sea Haven fork identity: commits AND PRs come in as our GitHub App bot
# `seahaven-openswe[bot]` (user id 296972425) so Slack/dashboard/schedule runs are
# attributed to the app, not the triggering user (deterministic; eliminates the
# self-review 422). The numeric noreply is the canonical GitHub form.
# NOTE (Adam, 2026-06-29 — ACCEPTED RISK): the `<id>+<login>@users.noreply` form
# may NOT resolve to a GitHub account Vercel preview deploys accept (the upstream
# `open-swe[bot]` hit exactly this and worked around it with a non-numeric
# address). We deliberately accept that risk here in exchange for a consistent
# bot identity across commits + PRs. If Vercel preview deploys on a target repo
# start rejecting our commits, this is the cause — revert to a resolvable address.
OPEN_SWE_BOT_NAME = "seahaven-openswe[bot]"
OPEN_SWE_BOT_EMAIL = "296972425+seahaven-openswe[bot]@users.noreply.github.com"


@dataclass(frozen=True)
class CollaboratorIdentity:
    """Identity used for git trailers and PR attribution."""

    display_name: str
    commit_name: str
    commit_email: str
    github_login: str = ""

    @property
    def pr_attribution_name(self) -> str:
        """Display name with GitHub login when available."""
        if self.github_login and self.github_login != self.display_name:
            return f"{self.display_name} (@{self.github_login})"
        return self.display_name


def _normalize_text(value: Any) -> str:
    return value.strip() if isinstance(value, str) else ""


def _github_noreply_email(login: str, user_id: Any = None) -> str:
    normalized_login = _normalize_text(login)
    if not normalized_login:
        return ""

    normalized_user_id = str(user_id).strip() if user_id is not None else ""
    if normalized_user_id:
        return f"{normalized_user_id}+{normalized_login}@users.noreply.github.com"
    return f"{normalized_login}@users.noreply.github.com"


def _identity_from_github_token(github_token: str | None) -> CollaboratorIdentity | None:
    if not github_token:
        return None

    try:
        response = httpx.get(
            "https://api.github.com/user",
            headers={
                "Authorization": f"Bearer {github_token}",
                "Accept": "application/vnd.github+json",
                "X-GitHub-Api-Version": "2022-11-28",
            },
            timeout=5.0,
        )
        if response.status_code != 200:  # noqa: PLR2004
            logger.debug("GitHub user lookup returned %s", response.status_code)
            return None

        payload = response.json()
        login = _normalize_text(payload.get("login"))
        display_name = _normalize_text(payload.get("name")) or login
        commit_email = _github_noreply_email(login, payload.get("id")) or _normalize_text(
            payload.get("email")
        )
        if not display_name or not commit_email:
            return None
        if commit_email == OPEN_SWE_BOT_EMAIL and display_name == OPEN_SWE_BOT_NAME:
            return None
        return CollaboratorIdentity(
            display_name=display_name,
            commit_name=display_name,
            commit_email=commit_email,
            github_login=login,
        )
    except httpx.HTTPError:
        logger.debug("Failed to resolve GitHub user identity from token", exc_info=True)
        return None


def _identity_from_config(config: dict[str, Any]) -> CollaboratorIdentity | None:
    configurable = config.get("configurable", {})
    slack_thread = configurable.get("slack_thread", {})
    linear_issue = configurable.get("linear_issue", {})

    display_name = (
        _normalize_text(slack_thread.get("triggering_user_name"))
        or _normalize_text(linear_issue.get("triggering_user_name"))
        or _normalize_text(configurable.get("user_email")).split("@", 1)[0]
    )

    github_login = _normalize_text(configurable.get("github_login"))
    if github_login:
        github_user_id = configurable.get("github_user_id")
        from ..dashboard.user_mappings import cached_email_for_login

        commit_email = _github_noreply_email(github_login, github_user_id) or _normalize_text(
            cached_email_for_login(github_login)
        )
        if commit_email:
            commit_name = display_name or github_login
            return CollaboratorIdentity(
                display_name=commit_name,
                commit_name=commit_name,
                commit_email=commit_email,
                github_login=github_login,
            )
    commit_email = _normalize_text(configurable.get("user_email")) or _normalize_text(
        slack_thread.get("triggering_user_email")
    )
    if display_name and commit_email:
        return CollaboratorIdentity(
            display_name=display_name,
            commit_name=display_name,
            commit_email=commit_email,
        )
    return None


def resolve_triggering_user_identity(
    config: dict[str, Any],
    github_token: str | None = None,
) -> CollaboratorIdentity | None:
    """Resolve the triggering user's git identity.

    Prefer the GitHub account identity derived from the token when available.
    Fall back to config metadata when the run originated from GitHub or when
    Slack/Linear supplied an explicit user name and email.
    """

    return _identity_from_github_token(github_token) or _identity_from_config(config)