open-swe/agent/webhooks/slack.py

"""Slack webhook handler — moved out of common.py (behavior-identical).

Helpers and constants stay in common.py; they are accessed through the module
object (``common.X``) so tests that monkeypatch them keep working.
"""

from datetime import UTC, datetime
from typing import Any

import httpx
from langchain_core.messages.content import create_text_block

from . import common


def _format_slack_thread_section(
    channel_id: str,
    thread_ts: str,
    context_source: str,
    channel_context: dict[str, Any] | None,
) -> str:
    lines = ["## Slack Thread", f"- Channel ID: {channel_id}"]
    channel_name = ""
    if isinstance(channel_context, dict):
        for key in ("name_normalized", "name"):
            value = channel_context.get(key)
            if isinstance(value, str) and value.strip():
                channel_name = value.strip()
                break
    if channel_name:
        lines.append(f"- Channel name: #{channel_name}")
    lines.append(f"- Thread TS: {thread_ts}")
    lines.append(f"- Context starts at: {context_source}")
    channel_description = common.get_slack_channel_context_description(channel_context)
    if channel_description:
        lines.extend(common.format_untrusted_channel_description(channel_description))
    return "\n".join(lines)


async def process_slack_mention(event_data: dict[str, Any], repo_config: dict[str, str]) -> None:
    """Process a Slack app mention by creating a run or queuing a mid-run message."""
    try:
        await _process_slack_mention_impl(event_data, repo_config)
    except Exception:  # noqa: BLE001
        common.logger.exception("Unexpected error while processing Slack mention")
        await _notify_slack_processing_error(event_data, repo_config)


async def _notify_slack_processing_error(
    event_data: dict[str, Any], repo_config: dict[str, str]
) -> None:
    channel_id = event_data.get("channel_id", "")
    thread_ts = event_data.get("thread_ts", "")
    event_ts = event_data.get("event_ts", "")
    user_id = event_data.get("user_id", "")
    text = event_data.get("text", "")
    bot_user_id = event_data.get("bot_user_id", "")
    if not channel_id or not thread_ts:
        return

    thread_id = common.generate_thread_id_from_slack_thread(channel_id, thread_ts)
    try:
        clean_text = (
            common.strip_bot_mention(text, bot_user_id, bot_username=common.SLACK_BOT_USERNAME)
            or "Slack request"
        )
        await common.upsert_agent_thread_owner_metadata(
            thread_id,
            source="slack",
            repo_config=repo_config,
            title=clean_text,
            source_context={
                "slack_thread": {
                    "channel_id": channel_id,
                    "thread_ts": thread_ts,
                    "triggering_user_id": user_id,
                    "triggering_event_ts": event_ts,
                }
            },
        )
    except Exception:  # noqa: BLE001
        common.logger.warning(
            "Could not persist Slack error metadata for thread %s", thread_id, exc_info=True
        )

    try:
        await common.get_client(url=common.LANGGRAPH_URL).threads.update(
            thread_id=thread_id,
            metadata={
                "latest_run_status": "error",
                "updated_at_ms": int(datetime.now(UTC).timestamp() * 1000),
            },
        )
    except Exception:  # noqa: BLE001
        common.logger.warning("Could not mark Slack thread %s as errored", thread_id, exc_info=True)

    try:
        await common.set_slack_assistant_status(channel_id, thread_ts, status="")
    except Exception:  # noqa: BLE001
        common.logger.debug("Could not clear Slack assistant status", exc_info=True)

    dashboard_url = common.dashboard_thread_url(thread_id)
    message = (
        "⚠️ I hit an unexpected error while handling this Slack thread. "
        "Send another message and I'll try again."
    )
    if dashboard_url:
        message += f" You can view the error in <{dashboard_url}|Open SWE Web>."
    try:
        await common.post_slack_thread_reply(channel_id, thread_ts, message)
    except Exception:  # noqa: BLE001
        common.logger.warning(
            "Could not post Slack error notification for thread %s", thread_id, exc_info=True
        )


async def _process_slack_mention_impl(
    event_data: dict[str, Any], repo_config: dict[str, str]
) -> None:
    channel_id = event_data.get("channel_id", "")
    thread_ts = event_data.get("thread_ts", "")
    event_ts = event_data.get("event_ts", "")
    user_id = event_data.get("user_id", "")
    text = event_data.get("text", "")
    bot_user_id = event_data.get("bot_user_id", "")
    channel_context_raw = event_data.get("channel_context")
    channel_context = (
        channel_context_raw
        if isinstance(channel_context_raw, dict)
        else common.normalize_slack_channel_context(channel_id, None)
    )

    if not channel_id or not thread_ts or not event_ts:
        common.logger.warning(
            "Missing Slack event fields (channel_id=%s, thread_ts=%s, event_ts=%s)",
            channel_id,
            thread_ts,
            event_ts,
        )
        return

    await common.set_slack_assistant_status(channel_id, thread_ts)

    thread_id = common.generate_thread_id_from_slack_thread(channel_id, thread_ts)

    # Prime the user-mapping cache so login/email/slack-id lookups below are warm.
    try:
        await common.refresh_user_mapping_cache()
    except Exception:  # noqa: BLE001
        common.logger.debug("Could not refresh user mapping cache for Slack mention", exc_info=True)

    user_email = None
    user_name = ""
    if user_id:
        slack_user = await common.get_slack_user_info(user_id)
        if slack_user:
            profile = slack_user.get("profile", {})
            if isinstance(profile, dict):
                user_email = profile.get("email")
                user_name = (
                    profile.get("display_name")
                    or profile.get("real_name")
                    or slack_user.get("real_name")
                    or slack_user.get("name")
                    or ""
                )

    thread_messages = await common.fetch_slack_thread_messages(channel_id, thread_ts)
    if not any(str(message.get("ts")) == str(event_ts) for message in thread_messages):
        thread_messages.append({"ts": event_ts, "text": text, "user": user_id})

    context_messages, context_mode = common.select_slack_context_messages(
        thread_messages, event_ts, bot_user_id, common.SLACK_BOT_USERNAME
    )
    context_user_ids = [
        value
        for value in (message.get("user") for message in context_messages)
        if isinstance(value, str) and value
    ]
    user_names_by_id = await common.get_slack_user_names(context_user_ids)
    if user_id and user_name and user_id not in user_names_by_id:
        user_names_by_id[user_id] = user_name
    context_text = common.format_slack_messages_for_prompt(
        context_messages,
        user_names_by_id,
        bot_user_id=bot_user_id,
        bot_username=common.SLACK_BOT_USERNAME,
    )
    context_source = (
        "the previous message where I was tagged"
        if context_mode == "last_mention"
        else "the beginning of the thread"
    )
    clean_text = (
        common.strip_bot_mention(text, bot_user_id, bot_username=common.SLACK_BOT_USERNAME)
        or "(no text in mention)"
    )
    trigger_user = user_name or (f"<@{user_id}>" if user_id else "Unknown user")

    # Auto-resolve cross-posted Slack message links in context
    resolved_links_section, image_urls_from_links = await common.resolve_slack_links_in_context(
        context_messages, user_names_by_id
    )

    slack_thread_section = _format_slack_thread_section(
        channel_id, thread_ts, context_source, channel_context
    )
    prompt = (
        "You were mentioned in Slack.\n\n"
        "## Default Repository Hint\n"
        f"{repo_config.get('owner')}/{repo_config.get('name')}\n"
        "Use this only if the Slack conversation does not identify a different repository.\n\n"
        f"## Triggered by\n{trigger_user}\n\n"
        f"{slack_thread_section}\n\n"
        f"## Conversation Context\n{context_text}\n\n"
        f"## Latest Mention Request\n{clean_text}\n\n"
        + (f"{resolved_links_section}\n\n" if resolved_links_section else "")
        + "Use `slack_thread_reply` to communicate in this Slack thread for clarifications, "
        "substantive updates, and final summaries. Use `slack_add_reaction` with :eyes: "
        "instead of posting perfunctory confirmation replies to user follow-up requests. "
        "Use `slack_read_thread_messages` to read any Slack messages by providing channel_id "
        "and message_ts."
    )
    content_blocks: list[dict[str, Any]] = [create_text_block(prompt)]

    image_urls = common.dedupe_urls(
        [url for msg in context_messages for url in common.extract_image_urls(msg.get("text", ""))]
        + [
            f["url_private"]
            for msg in context_messages
            for f in msg.get("files", [])
            if isinstance(f, dict)
            and f.get("mimetype", "").startswith("image/")
            and f.get("url_private")
        ]
        + image_urls_from_links
    )

    mapped_login = await common.login_for_slack_id(user_id)
    if not mapped_login and user_email:
        mapped_login = await common.login_for_email(user_email)

    image_model_override: tuple[str, str] | None = None
    if image_urls:
        resolved_model_id = await common.resolve_agent_model_id(mapped_login)
        if not common.model_supports_images(resolved_model_id):
            fallback_model_id, fallback_effort = common.default_vision_model_pair()
            common.logger.info(
                "Using vision fallback model %s for %d Slack image(s); configured model %s "
                "does not support images",
                fallback_model_id,
                len(image_urls),
                resolved_model_id,
            )
            resolved_model_id = fallback_model_id
            image_model_override = (fallback_model_id, fallback_effort)
        common.logger.info("Preparing %d image(s) for Slack mention", len(image_urls))
        async with httpx.AsyncClient(timeout=common.DEFAULT_HTTP_TIMEOUT) as http_client:
            for image_url in image_urls:
                image_block = await common.fetch_image_block(image_url, http_client)
                if image_block:
                    content_blocks.append(image_block)

    # Open SWE opens PRs as the triggering user, so a run only proceeds when we
    # have a valid user GitHub token. Users who have never signed in with
    # GitHub, and users whose stored authorization is no longer usable, are
    # blocked and prompted to set up via the dashboard. Bot-token-only
    # deployments are exempt — they run on the installation token.
    user_token: str | None = None
    if mapped_login:
        try:
            user_token = await common.get_valid_access_token(mapped_login)
        except Exception:  # noqa: BLE001
            common.logger.debug(
                "Failed to resolve GitHub token for %s; treating as unauthenticated",
                mapped_login,
                exc_info=True,
            )
            user_token = None
    has_valid_user_token = bool(user_token)

    if not has_valid_user_token and not common.is_bot_token_only_mode():
        # A stored-but-unusable token means "sign in again"; no record at all
        # means the user has never connected GitHub + Slack via the dashboard.
        # Guard the store read like token resolution above so a transient
        # failure still yields an actionable prompt and clears the status.
        has_token_record = False
        if mapped_login:
            try:
                has_token_record = await common.has_access_token_record(mapped_login)
            except Exception:  # noqa: BLE001
                common.logger.debug(
                    "Failed to check GitHub token record for %s; prompting sign-in",
                    mapped_login,
                    exc_info=True,
                )
        reason = "revoked" if has_token_record else "unlinked"
        common.logger.info(
            "Blocking Slack run for thread %s: no valid user GitHub token (%s)",
            thread_id,
            reason,
        )
        if user_id:
            await common._post_account_link_prompt(
                channel_id, thread_ts, user_id, user_email, reason=reason
            )
        await common.set_slack_assistant_status(channel_id, thread_ts, status="")
        return

    configurable: dict[str, Any] = {
        "repo": repo_config,
        "slack_thread": {
            "channel_id": channel_id,
            "channel_context": channel_context,
            "thread_ts": thread_ts,
            "triggering_user_id": user_id,
            "triggering_user_name": user_name,
            "triggering_user_email": user_email,
            "triggering_event_ts": event_ts,
        },
        "user_email": user_email,
        "source": "slack",
    }
    if mapped_login:
        configurable["github_login"] = mapped_login
    if image_model_override:
        configurable["agent_model_id"] = image_model_override[0]
        configurable["agent_effort"] = image_model_override[1]

    thread_plan_mode = await common._get_thread_plan_mode(thread_id)
    if thread_plan_mode is not None:
        configurable["plan_mode"] = thread_plan_mode

    langgraph_client = common.get_client(url=common.LANGGRAPH_URL)
    is_first_mention = not await common._thread_exists(thread_id)
    await common._upsert_slack_thread_repo_metadata(thread_id, repo_config, langgraph_client)
    # Pass the login resolved above (from the stable Slack user id) so the thread is
    # always tagged with github_login — the key the dashboard searches by. Without
    # it, upsert re-resolves from the Slack profile email, which can miss.
    await common.upsert_agent_thread_owner_metadata(
        thread_id,
        source="slack",
        repo_config=repo_config,
        github_login=mapped_login or "",
        user_email=user_email or "",
        title=clean_text if is_first_mention else "",
        source_context={"slack_thread": configurable["slack_thread"]},
    )

    run = await common.dispatch_agent_run(
        thread_id,
        content_blocks,
        configurable,
        source="slack",
        metadata=common._AGENT_VERSION_METADATA,
        client=langgraph_client,
    )
    common.logger.info(
        "Slack LangGraph run %s dispatched for thread %s",
        common._run_id_for_logging(run),
        thread_id,
    )
    run_id = run.get("run_id")
    if is_first_mention:
        trace_message_ts = await common.post_slack_trace_reply(channel_id, thread_ts, thread_id)
        await common.set_slack_assistant_status(channel_id, thread_ts)
        if isinstance(run_id, str) and run_id:
            await common.store_slack_run_mapping(
                langgraph_client,
                channel_id,
                thread_ts,
                run_id,
                message_ts=trace_message_ts,
                trace_message_ts=trace_message_ts,
                triggering_user_id=user_id,
            )
    else:
        common.logger.info(
            "Skipping Slack trace reply for thread %s — agent will reply when run completes",
            thread_id,
        )
        if isinstance(run_id, str) and run_id:
            await common.store_slack_run_mapping(
                langgraph_client,
                channel_id,
                thread_ts,
                run_id,
                triggering_user_id=user_id,
            )