diff --git a/agent/prompt.py b/agent/prompt.py index c415e1ec..b6192e2c 100644 --- a/agent/prompt.py +++ b/agent/prompt.py @@ -83,6 +83,7 @@ OPEN_SWE_SHARED_BASE = """You are **Open SWE**, an open-source agent built on La ### Communication - Focus on the substance and keep summaries brief. Use light markdown (`###`/`####` headings, bold, code) — avoid `#`/`##` titles. +- In Slack, when a user asks to "break out," "split out," or "start a separate thread" for part of the work, summarize the requested aspect and relevant context into self-contained instructions, then call `slack_start_new_thread` instead of only replying in the current thread. - When you post to Slack with `slack_thread_reply`, do not repeat that text in a later assistant message; the user can already see the Slack message. - When delegated work to a subagent: the calling agent only sees your final message, so make it the complete answer. diff --git a/agent/server.py b/agent/server.py index 12a6fa80..c16d36ab 100644 --- a/agent/server.py +++ b/agent/server.py @@ -86,6 +86,7 @@ from .tools import ( save_plan, schedule_thread_wakeup, slack_read_thread_messages, + slack_start_new_thread, slack_thread_reply, web_search, ) @@ -624,6 +625,7 @@ PLAN_MODE_EXCLUDED_TOOLS: frozenset[str] = frozenset( "http_request", "open_pull_request", "request_pr_review", + "slack_start_new_thread", "linear_create_issue", "linear_update_issue", "linear_delete_issue", @@ -958,6 +960,7 @@ async def get_agent(config: RunnableConfig) -> Pregel: request_pr_review, schedule_thread_wakeup, slack_read_thread_messages, + slack_start_new_thread, slack_thread_reply, *corridor_tools, *observability_tools, diff --git a/agent/tools/__init__.py b/agent/tools/__init__.py index ef411c17..532324d2 100644 --- a/agent/tools/__init__.py +++ b/agent/tools/__init__.py @@ -21,6 +21,7 @@ from .save_plan import save_plan from .schedule_thread_wakeup import schedule_thread_wakeup from .search_repo_code import search_repo_code from .slack_read_thread_messages import slack_read_thread_messages +from .slack_start_new_thread import slack_start_new_thread from .slack_thread_reply import slack_thread_reply from .update_finding import update_finding from .web_search import web_search @@ -49,6 +50,7 @@ __all__ = [ "schedule_thread_wakeup", "search_repo_code", "slack_read_thread_messages", + "slack_start_new_thread", "slack_thread_reply", "update_finding", "web_search", diff --git a/agent/tools/slack_start_new_thread.py b/agent/tools/slack_start_new_thread.py new file mode 100644 index 00000000..cf9de0ce --- /dev/null +++ b/agent/tools/slack_start_new_thread.py @@ -0,0 +1,260 @@ +import os +import re +from typing import Any + +from langgraph.config import get_config +from langgraph_sdk import get_client + +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 + +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", + } + + 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), + } diff --git a/tests/test_plan_mode.py b/tests/test_plan_mode.py index cc3ea0ae..cbf8ca59 100644 --- a/tests/test_plan_mode.py +++ b/tests/test_plan_mode.py @@ -29,6 +29,7 @@ def test_plan_mode_excluded_tools_cover_mutating_tools() -> None: "task", "open_pull_request", "request_pr_review", + "slack_start_new_thread", "linear_create_issue", "linear_update_issue", "linear_delete_issue", diff --git a/tests/test_slack_start_new_thread_tool.py b/tests/test_slack_start_new_thread_tool.py new file mode 100644 index 00000000..b7edf311 --- /dev/null +++ b/tests/test_slack_start_new_thread_tool.py @@ -0,0 +1,259 @@ +from __future__ import annotations + +import importlib +from typing import Any + +import pytest + +from agent.utils.thread_ids import generate_thread_id_from_slack_thread + +slack_breakout_tool = importlib.import_module("agent.tools.slack_start_new_thread") + + +def _config() -> dict[str, Any]: + return { + "configurable": { + "repo": {"owner": "langchain-ai", "name": "open-swe"}, + "github_login": "alice", + "user_email": "alice@example.com", + "agent_model_id": "bedrock_converse:us.anthropic.claude-opus-4-8", + "agent_effort": "high", + "slack_thread": { + "channel_id": "C1", + "thread_ts": "1700000000.000001", + "triggering_user_id": "U1", + "triggering_user_name": "Alice", + "triggering_user_email": "alice@example.com", + "triggering_event_ts": "1700000000.000002", + }, + } + } + + +class _FakeThreadsClient: + def __init__(self, captured: dict[str, Any]) -> None: + self.captured = captured + + async def create(self, *, thread_id: str, if_exists: str, metadata: dict[str, Any]) -> None: + self.captured["thread_create"] = { + "thread_id": thread_id, + "if_exists": if_exists, + "metadata": metadata, + } + + async def update(self, *, thread_id: str, metadata: dict[str, Any]) -> None: + self.captured["thread_update"] = {"thread_id": thread_id, "metadata": metadata} + + +class _FakeClient: + def __init__(self, captured: dict[str, Any]) -> None: + self.threads = _FakeThreadsClient(captured) + + +async def test_slack_start_new_thread_success(monkeypatch: pytest.MonkeyPatch) -> None: + captured: dict[str, Any] = {"stored_mappings": []} + new_ts = "1700000000.111111" + trace_ts = "1700000000.222222" + + async def fake_post_top_level( + channel_id: str, + text: str, + *, + unfurl_links: bool = True, + unfurl_media: bool = True, + blocks: list[dict[str, Any]] | None = None, + ) -> tuple[str | None, str | None]: + captured["top_level_post"] = { + "channel_id": channel_id, + "text": text, + "unfurl_links": unfurl_links, + "unfurl_media": unfurl_media, + "blocks": blocks, + } + return new_ts, None + + async def fake_dispatch_agent_run( + thread_id: str, + content: str, + configurable: dict[str, Any], + *, + source: str, + client: Any, + **kwargs: Any, + ) -> dict[str, str]: + captured["dispatch"] = { + "thread_id": thread_id, + "content": content, + "configurable": configurable, + "source": source, + "client": client, + "kwargs": kwargs, + } + return {"run_id": "run-123"} + + async def fake_post_trace(channel_id: str, thread_ts: str, thread_id: str) -> str: + captured["trace"] = { + "channel_id": channel_id, + "thread_ts": thread_ts, + "thread_id": thread_id, + } + return trace_ts + + async def fake_store_mapping( + client: Any, + channel_id: str, + thread_ts: str, + run_id: str, + *, + message_ts: str | None = None, + triggering_user_id: str | None = None, + ) -> None: + captured["stored_mappings"].append( + { + "client": client, + "channel_id": channel_id, + "thread_ts": thread_ts, + "run_id": run_id, + "message_ts": message_ts, + "triggering_user_id": triggering_user_id, + } + ) + + fake_client = _FakeClient(captured) + monkeypatch.setattr(slack_breakout_tool, "get_config", _config) + monkeypatch.setattr(slack_breakout_tool, "get_client", lambda url: fake_client) + monkeypatch.setattr( + slack_breakout_tool, "post_slack_top_level_message_with_ts", fake_post_top_level + ) + monkeypatch.setattr(slack_breakout_tool, "dispatch_agent_run", fake_dispatch_agent_run) + monkeypatch.setattr(slack_breakout_tool, "post_slack_trace_reply", fake_post_trace) + monkeypatch.setattr(slack_breakout_tool, "store_slack_run_mapping", fake_store_mapping) + monkeypatch.setattr( + slack_breakout_tool, + "dashboard_thread_url", + lambda thread_id: f"https://dashboard.example/agents/{thread_id}", + ) + + result = await slack_breakout_tool.slack_start_new_thread( + "Investigate follow-up", + "Use the same repo and investigate the follow-up aspect in detail.", + ) + + expected_thread_id = generate_thread_id_from_slack_thread("C1", new_ts) + assert result == { + "success": True, + "thread_id": expected_thread_id, + "thread_ts": new_ts, + "dashboard_url": f"https://dashboard.example/agents/{expected_thread_id}", + } + assert captured["top_level_post"]["channel_id"] == "C1" + assert "Investigate follow-up" in captured["top_level_post"]["text"] + assert "langchain-ai/open-swe" in captured["top_level_post"]["text"] + assert captured["top_level_post"]["unfurl_links"] is False + assert captured["thread_create"]["if_exists"] == "do_nothing" + assert captured["thread_create"]["thread_id"] == expected_thread_id + metadata = captured["thread_update"]["metadata"] + assert metadata["source"] == "slack" + assert metadata["repo"] == {"owner": "langchain-ai", "name": "open-swe"} + assert metadata["github_login"] == "alice" + assert metadata["triggering_user_email"] == "alice@example.com" + assert metadata["source_context"]["slack_thread"]["thread_ts"] == new_ts + assert metadata["source_context"]["slack_thread"]["triggering_user_id"] == "U1" + assert metadata["source_context"]["breakout_from"] == { + "channel_id": "C1", + "thread_ts": "1700000000.000001", + "message_ts": "1700000000.000002", + } + dispatch = captured["dispatch"] + assert dispatch["thread_id"] == expected_thread_id + assert dispatch["source"] == "slack" + assert dispatch["configurable"]["slack_thread"]["thread_ts"] == new_ts + assert dispatch["configurable"]["repo"] == {"owner": "langchain-ai", "name": "open-swe"} + assert dispatch["configurable"]["github_login"] == "alice" + assert ( + dispatch["configurable"]["agent_model_id"] + == "bedrock_converse:us.anthropic.claude-opus-4-8" + ) + assert "Breakout Instructions" in dispatch["content"] + assert captured["trace"] == { + "channel_id": "C1", + "thread_ts": new_ts, + "thread_id": expected_thread_id, + } + assert [item["message_ts"] for item in captured["stored_mappings"]] == [new_ts, trace_ts] + assert all(item["triggering_user_id"] == "U1" for item in captured["stored_mappings"]) + + +async def test_slack_start_new_thread_requires_slack_config( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setattr(slack_breakout_tool, "get_config", lambda: {"configurable": {}}) + + result = await slack_breakout_tool.slack_start_new_thread("Title", "Instructions") + + assert result == {"success": False, "error": "Missing slack_thread config"} + + +@pytest.mark.parametrize( + ("title", "instructions", "error"), + [ + ("", "Instructions", "title is required"), + ("Title", "", "instructions is required"), + ("x" * 161, "Instructions", "title is too long"), + ("Title", "x" * 12001, "instructions is too long"), + ], +) +async def test_slack_start_new_thread_validates_text( + title: str, + instructions: str, + error: str, + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setattr(slack_breakout_tool, "get_config", _config) + + result = await slack_breakout_tool.slack_start_new_thread(title, instructions) + + assert result["success"] is False + assert result["error"] == error + + +async def test_slack_start_new_thread_rejects_invalid_repo_override( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setattr(slack_breakout_tool, "get_config", _config) + + result = await slack_breakout_tool.slack_start_new_thread( + "Title", "Instructions", default_repo="https://github.com/langchain-ai/open-swe" + ) + + assert result == { + "success": False, + "error": "default_repo must be a simple owner/name repository string", + } + + +async def test_slack_start_new_thread_returns_slack_failure_without_dispatch( + monkeypatch: pytest.MonkeyPatch, +) -> None: + captured: dict[str, bool] = {"dispatched": False} + + async def fake_post_top_level(*args: Any, **kwargs: Any) -> tuple[str | None, str | None]: + return None, "msg_too_long" + + async def fake_dispatch_agent_run(*args: Any, **kwargs: Any) -> dict[str, str]: + captured["dispatched"] = True + return {"run_id": "run-123"} + + monkeypatch.setattr(slack_breakout_tool, "get_config", _config) + monkeypatch.setattr( + slack_breakout_tool, "post_slack_top_level_message_with_ts", fake_post_top_level + ) + monkeypatch.setattr(slack_breakout_tool, "dispatch_agent_run", fake_dispatch_agent_run) + + result = await slack_breakout_tool.slack_start_new_thread("Title", "Instructions") + + assert result["success"] is False + assert result["error"] == "msg_too_long" + assert result["slack_error"] == "msg_too_long" + assert "shorter" in result["hint"] + assert captured["dispatched"] is False