open-swe/agent/tools/slack_start_new_thread.py

import os
import re
from typing import Any

from fastapi import HTTPException
from langgraph.config import get_config
from langgraph_sdk import get_client

from ..dashboard.repo_access import require_repo_access_for_user
from ..dispatch import dispatch_agent_run
from ..utils.dashboard_links import dashboard_thread_url
from ..utils.slack import (
    post_slack_top_level_message_with_ts,
    post_slack_trace_reply,
    store_slack_run_mapping,
)
from ..utils.thread_ids import generate_thread_id_from_slack_thread
from ..webhooks.common import _is_repo_allowed

LANGGRAPH_URL = os.environ.get("LANGGRAPH_URL") or os.environ.get(
    "LANGGRAPH_URL_PROD", "http://localhost:2024"
)

_TITLE_MAX_CHARS = 160
_INSTRUCTIONS_MAX_CHARS = 12000
_VISIBLE_INSTRUCTIONS_MAX_CHARS = 2800
_REPO_RE = re.compile(r"^[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+$")


def _failure_hint(slack_error: str | None) -> str:
    if slack_error == "msg_too_long":
        return "Slack rejected the message as too long; retry with shorter title or instructions."
    if slack_error in {"channel_not_found", "not_in_channel"}:
        return "Slack rejected the channel; do not retry with another channel."
    if slack_error and slack_error.startswith("rate_limited"):
        retry_after = slack_error.partition(":")[2].strip()
        if retry_after:
            return f"Slack rate limited the request; wait at least {retry_after}s before retrying."
        return "Slack rate limited the request; wait before retrying."
    if slack_error == "missing_slack_bot_token":
        return "Slack bot token is missing; do not retry."
    if slack_error and slack_error.startswith("http_error:"):
        return "Slack posting hit an HTTP error; retry once."
    return "Slack post failed; retry once with concise instructions."


def _validate_text(value: str, *, field: str, max_chars: int) -> str | dict[str, Any]:
    text = value.strip() if isinstance(value, str) else ""
    if not text:
        return {"success": False, "error": f"{field} is required"}
    if len(text) > max_chars:
        return {
            "success": False,
            "error": f"{field} is too long",
            "max_chars": max_chars,
            "actual_chars": len(text),
        }
    return text


def _resolve_repo(configurable: dict[str, Any], default_repo: str | None) -> dict[str, str] | None:
    if default_repo and default_repo.strip():
        candidate = default_repo.strip()
        if not _REPO_RE.fullmatch(candidate):
            return None
        owner, name = candidate.split("/", 1)
        return {"owner": owner, "name": name}

    repo = configurable.get("repo")
    if isinstance(repo, dict):
        owner = repo.get("owner")
        name = repo.get("name")
        if isinstance(owner, str) and owner.strip() and isinstance(name, str) and name.strip():
            return {"owner": owner.strip(), "name": name.strip()}
    return None


def _truncate_for_slack(text: str) -> str:
    if len(text) <= _VISIBLE_INSTRUCTIONS_MAX_CHARS:
        return text
    omitted = len(text) - _VISIBLE_INSTRUCTIONS_MAX_CHARS
    return f"{text[:_VISIBLE_INSTRUCTIONS_MAX_CHARS].rstrip()}\n\n…truncated {omitted} chars; the new Open SWE thread received the full instructions."


def _visible_message(title: str, instructions: str, repo: dict[str, str] | None) -> str:
    repo_line = f"\n*Repository:* `{repo['owner']}/{repo['name']}`" if repo else ""
    return (
        f"*Open SWE breakout thread:* {title}{repo_line}\n\n"
        f"*Instructions for the new thread:*\n{_truncate_for_slack(instructions)}"
    )


def _run_prompt(
    title: str,
    instructions: str,
    repo: dict[str, str] | None,
    original_slack_thread: dict[str, Any],
) -> str:
    repo_text = f"{repo['owner']}/{repo['name']}" if repo else "(no repository specified)"
    channel_id = original_slack_thread.get("channel_id", "")
    thread_ts = original_slack_thread.get("thread_ts", "")
    return (
        "You were started from another Open SWE Slack thread as a breakout task.\n\n"
        f"## Breakout Title\n{title}\n\n"
        f"## Default Repository Hint\n{repo_text}\n"
        "Use this repository unless the instructions below clearly identify a different repository.\n\n"
        "## Source Slack Thread\n"
        f"- Channel: {channel_id}\n"
        f"- Thread TS: {thread_ts}\n\n"
        "## Breakout Instructions\n"
        f"{instructions}\n\n"
        "Use `slack_thread_reply` to communicate in this new Slack thread for clarifications, "
        "status updates, and final summaries."
    )


def _new_slack_thread_context(
    original: dict[str, Any],
    *,
    channel_id: str,
    thread_ts: str,
) -> dict[str, Any]:
    return {
        "channel_id": channel_id,
        "thread_ts": thread_ts,
        "triggering_user_id": original.get("triggering_user_id", ""),
        "triggering_user_name": original.get("triggering_user_name", ""),
        "triggering_user_email": original.get("triggering_user_email", ""),
        "triggering_event_ts": thread_ts,
    }


async def slack_start_new_thread(
    title: str,
    instructions: str,
    default_repo: str | None = None,
) -> dict[str, Any]:
    """Start a new Open SWE thread in a top-level Slack message in the current channel."""
    config = get_config()
    configurable = config.get("configurable", {})
    current_slack_thread = configurable.get("slack_thread")
    if not isinstance(current_slack_thread, dict):
        return {"success": False, "error": "Missing slack_thread config"}

    channel_id = current_slack_thread.get("channel_id")
    current_thread_ts = current_slack_thread.get("thread_ts")
    if not isinstance(channel_id, str) or not channel_id.strip():
        return {"success": False, "error": "Missing slack_thread.channel_id in config"}

    clean_title = _validate_text(title, field="title", max_chars=_TITLE_MAX_CHARS)
    if isinstance(clean_title, dict):
        return clean_title
    clean_instructions = _validate_text(
        instructions, field="instructions", max_chars=_INSTRUCTIONS_MAX_CHARS
    )
    if isinstance(clean_instructions, dict):
        return clean_instructions

    repo = _resolve_repo(configurable, default_repo)
    if default_repo and default_repo.strip() and repo is None:
        return {
            "success": False,
            "error": "default_repo must be a simple owner/name repository string",
        }

    if default_repo and default_repo.strip() and repo is not None:
        if not _is_repo_allowed(repo):
            return {
                "success": False,
                "error": (
                    f"Repository {repo['owner']}/{repo['name']} is not on the deployment allowlist"
                ),
            }
        github_login = configurable.get("github_login")
        if not isinstance(github_login, str) or not github_login.strip():
            return {
                "success": False,
                "error": (
                    "Cannot verify access to the requested repository: no github_login on the "
                    "parent thread"
                ),
            }
        try:
            await require_repo_access_for_user(
                github_login.strip(), f"{repo['owner']}/{repo['name']}"
            )
        except HTTPException as exc:
            return {
                "success": False,
                "error": (
                    f"Access to repository {repo['owner']}/{repo['name']} denied: {exc.detail}"
                ),
            }
        except Exception as exc:  # noqa: BLE001
            return {
                "success": False,
                "error": (
                    f"Failed to verify access to repository {repo['owner']}/{repo['name']}: {exc}"
                ),
            }

    message_ts, slack_error = await post_slack_top_level_message_with_ts(
        channel_id.strip(),
        _visible_message(clean_title, clean_instructions, repo),
        unfurl_links=False,
        unfurl_media=False,
    )
    if message_ts is None:
        return {
            "success": False,
            "error": slack_error or "post failed",
            "slack_error": slack_error,
            "hint": _failure_hint(slack_error),
        }

    thread_id = generate_thread_id_from_slack_thread(channel_id.strip(), message_ts)
    new_slack_thread = _new_slack_thread_context(
        current_slack_thread,
        channel_id=channel_id.strip(),
        thread_ts=message_ts,
    )
    breakout_from = {
        "channel_id": channel_id.strip(),
        "thread_ts": current_thread_ts or "",
        "message_ts": current_slack_thread.get("triggering_event_ts", ""),
    }

    metadata: dict[str, Any] = {
        "source": "slack",
        "title": clean_title[:80],
        "source_context": {
            "slack_thread": new_slack_thread,
            "breakout_from": breakout_from,
        },
    }
    if repo:
        metadata.update(
            {
                "repo": repo,
                "repo_owner": repo["owner"],
                "repo_name": repo["name"],
            }
        )
    github_login = configurable.get("github_login")
    if isinstance(github_login, str) and github_login:
        metadata["github_login"] = github_login
    user_email = configurable.get("user_email")
    if isinstance(user_email, str) and user_email:
        metadata["triggering_user_email"] = user_email.strip().lower()

    new_configurable: dict[str, Any] = {
        "slack_thread": new_slack_thread,
        "source": "slack",
    }
    if repo:
        new_configurable["repo"] = repo
    for key in ("user_email", "github_login", "agent_model_id", "agent_effort"):
        value = configurable.get(key)
        if value:
            new_configurable[key] = value

    client = get_client(url=LANGGRAPH_URL)
    await client.threads.create(thread_id=thread_id, if_exists="do_nothing", metadata=metadata)
    await client.threads.update(thread_id=thread_id, metadata=metadata)

    run = await dispatch_agent_run(
        thread_id,
        _run_prompt(clean_title, clean_instructions, repo, current_slack_thread),
        new_configurable,
        source="slack",
        client=client,
    )
    run_id = run.get("run_id") if isinstance(run, dict) else None
    trace_message_ts = await post_slack_trace_reply(channel_id.strip(), message_ts, thread_id)
    if isinstance(run_id, str) and run_id:
        await store_slack_run_mapping(
            client,
            channel_id.strip(),
            message_ts,
            run_id,
            message_ts=message_ts,
            triggering_user_id=new_slack_thread.get("triggering_user_id") or None,
        )
        if trace_message_ts:
            await store_slack_run_mapping(
                client,
                channel_id.strip(),
                message_ts,
                run_id,
                message_ts=trace_message_ts,
                triggering_user_id=new_slack_thread.get("triggering_user_id") or None,
            )

    return {
        "success": True,
        "thread_id": thread_id,
        "thread_ts": message_ts,
        "dashboard_url": dashboard_thread_url(thread_id),
    }