diff --git a/agent/prompt.py b/agent/prompt.py
index 4b4b65cd..0f35b1e8 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 944a2b6f..58571275 100644
--- a/agent/server.py
+++ b/agent/server.py
@@ -85,6 +85,7 @@ from .tools import (
     save_plan,
     schedule_thread_wakeup,
     slack_read_thread_messages,
+    slack_start_new_thread,
     slack_thread_reply,
     web_search,
 )
@@ -546,6 +547,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",
@@ -836,6 +838,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/agent/utils/slack.py b/agent/utils/slack.py
index 82e2626a..04134a53 100644
--- a/agent/utils/slack.py
+++ b/agent/utils/slack.py
@@ -297,26 +297,26 @@ async def set_slack_assistant_status(
             return False
 
 
-async def post_slack_thread_reply_with_ts(
+async def _post_slack_message_with_ts(
     channel_id: str,
-    thread_ts: str,
     text: str,
     *,
+    thread_ts: str | None = None,
     unfurl_links: bool = True,
     unfurl_media: bool = True,
     blocks: list[dict[str, Any]] | None = None,
 ) -> tuple[str | None, str | None]:
-    """Post a reply in a Slack thread and return its Slack timestamp and error."""
     if not SLACK_BOT_TOKEN:
         return None, "missing_slack_bot_token"
 
     payload: dict[str, Any] = {
         "channel": channel_id,
-        "thread_ts": thread_ts,
         "text": text,
         "unfurl_links": unfurl_links,
         "unfurl_media": unfurl_media,
     }
+    if thread_ts is not None:
+        payload["thread_ts"] = thread_ts
     if blocks:
         payload["blocks"] = blocks
 
@@ -350,6 +350,44 @@ async def post_slack_thread_reply_with_ts(
             return None, f"http_error: {type(exc).__name__}"
 
 
+async def post_slack_thread_reply_with_ts(
+    channel_id: str,
+    thread_ts: str,
+    text: str,
+    *,
+    unfurl_links: bool = True,
+    unfurl_media: bool = True,
+    blocks: list[dict[str, Any]] | None = None,
+) -> tuple[str | None, str | None]:
+    """Post a reply in a Slack thread and return its Slack timestamp and error."""
+    return await _post_slack_message_with_ts(
+        channel_id,
+        text,
+        thread_ts=thread_ts,
+        unfurl_links=unfurl_links,
+        unfurl_media=unfurl_media,
+        blocks=blocks,
+    )
+
+
+async def post_slack_top_level_message_with_ts(
+    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]:
+    """Post a top-level Slack message and return its timestamp and error."""
+    return await _post_slack_message_with_ts(
+        channel_id,
+        text,
+        unfurl_links=unfurl_links,
+        unfurl_media=unfurl_media,
+        blocks=blocks,
+    )
+
+
 async def update_slack_message(
     channel_id: str,
     message_ts: str,
diff --git a/agent/utils/thread_ids.py b/agent/utils/thread_ids.py
new file mode 100644
index 00000000..eba5a99a
--- /dev/null
+++ b/agent/utils/thread_ids.py
@@ -0,0 +1,9 @@
+import hashlib
+import uuid
+
+
+def generate_thread_id_from_slack_thread(channel_id: str, thread_ts: str) -> str:
+    """Generate a deterministic thread ID from a Slack thread identifier."""
+    composite = f"{channel_id}:{thread_ts}"
+    md5_hex = hashlib.md5(composite.encode("utf-8")).hexdigest()
+    return str(uuid.UUID(hex=md5_hex))
diff --git a/agent/webapp.py b/agent/webapp.py
index ea9259d5..98e38c8a 100644
--- a/agent/webapp.py
+++ b/agent/webapp.py
@@ -124,6 +124,7 @@ from .utils.slack_feedback import (
     process_slack_reaction_added,
     process_slack_reaction_removed,
 )
+from .utils.thread_ids import generate_thread_id_from_slack_thread
 
 logger = logging.getLogger(__name__)
 
@@ -377,13 +378,6 @@ def generate_thread_id_from_github_issue(issue_id: str) -> str:
     )
 
 
-def generate_thread_id_from_slack_thread(channel_id: str, thread_id: str) -> str:
-    """Generate a deterministic thread ID from a Slack thread identifier."""
-    composite = f"{channel_id}:{thread_id}"
-    md5_hex = hashlib.md5(composite.encode("utf-8")).hexdigest()
-    return str(uuid.UUID(hex=md5_hex))
-
-
 def generate_reviewer_thread_id(owner: str, repo: str, pr_number: int) -> str:
     stable_key = f"{owner}/{repo}/pr/{pr_number}/reviewer"
     return str(uuid.uuid5(uuid.NAMESPACE_URL, stable_key))
diff --git a/tests/e2e/fake_llm.py b/tests/e2e/fake_llm.py
index 18971b41..36e7941a 100644
--- a/tests/e2e/fake_llm.py
+++ b/tests/e2e/fake_llm.py
@@ -11,6 +11,8 @@ from __future__ import annotations
 
 import re
 import time
+from collections.abc import Callable
+from dataclasses import dataclass
 from typing import Any
 
 from e2e_env import (
@@ -46,8 +48,63 @@ git push origin {FEATURE_BRANCH}
 echo PUSHED_OK
 """.strip()
 
-
 _PLAN_URL_RE = re.compile(r"https?://[^\s\"'<>)\]|]+/plan\b")
+_ATTRIBUTION_RE = re.compile(r"@([A-Za-z0-9-]+):")
+
+ToolArgs = dict[str, Any]
+StepFactory = Callable[[list[BaseMessage]], AIMessage]
+ScriptPredicate = Callable[["ScriptContext"], bool]
+
+
+@dataclass(frozen=True)
+class ToolCallSpec:
+    name: str
+    args: ToolArgs
+    call_id: str
+
+
+@dataclass(frozen=True)
+class StepSpec:
+    content: str = ""
+    tool_calls: tuple[ToolCallSpec, ...] = ()
+    factory: StepFactory | None = None
+
+
+@dataclass(frozen=True)
+class ScriptContext:
+    first_text: str
+    last_text: str
+    human_count: int
+
+
+@dataclass(frozen=True)
+class ScriptRule:
+    name: str
+    predicate: ScriptPredicate
+
+
+def _tool_call(name: str, args: ToolArgs, call_id: str) -> ToolCallSpec:
+    return ToolCallSpec(name=name, args=args, call_id=call_id)
+
+
+def _tool_step(content: str, name: str, args: ToolArgs, call_id: str) -> StepSpec:
+    return StepSpec(content=content, tool_calls=(_tool_call(name, args, call_id),))
+
+
+def _dynamic_step(factory: StepFactory) -> StepSpec:
+    return StepSpec(factory=factory)
+
+
+def _render_step(step: StepSpec, messages: list[BaseMessage]) -> AIMessage:
+    if step.factory is not None:
+        return step.factory(messages)
+    return AIMessage(
+        content=step.content,
+        tool_calls=[
+            {"name": call.name, "args": dict(call.args), "id": call.call_id}
+            for call in step.tool_calls
+        ],
+    )
 
 
 def _text(content: Any) -> str:
@@ -71,8 +128,7 @@ def _pr_url_from_messages(messages: list[BaseMessage]) -> str | None:
 
 
 def _plan_url_from_messages(messages: list[BaseMessage]) -> str | None:
-    """The plan-review URL is injected into the system prompt; a real model would
-    read it the same way."""
+    """The plan-review URL is injected into the system prompt; a real model would read it."""
     for msg in messages:
         match = _PLAN_URL_RE.search(_text(msg.content))
         if match:
@@ -92,35 +148,7 @@ def _reviewer_feedback(messages: list[BaseMessage]) -> str | None:
     return None
 
 
-def _step_implement(_messages: list[BaseMessage]) -> AIMessage:
-    return AIMessage(
-        content="Setting up the repo and implementing the change.",
-        tool_calls=[{"name": "execute", "args": {"command": _IMPLEMENT_SCRIPT}, "id": "call-impl"}],
-    )
-
-
-def _step_open_pr(_messages: list[BaseMessage]) -> AIMessage:
-    return AIMessage(
-        content="Opening a pull request.",
-        tool_calls=[
-            {
-                "name": "open_pull_request",
-                "args": {
-                    "owner": OWNER,
-                    "repo": REPO,
-                    "head": FEATURE_BRANCH,
-                    "base": BASE_BRANCH,
-                    "title": PR_TITLE,
-                    "body": "Adds a `greet()` helper as requested.",
-                    "draft": True,
-                },
-                "id": "call-pr",
-            }
-        ],
-    )
-
-
-def _step_reply(messages: list[BaseMessage]) -> AIMessage:
+def _reply_step(messages: list[BaseMessage]) -> AIMessage:
     url = _pr_url_from_messages(messages) or "(PR url unavailable)"
     feedback = _reviewer_feedback(messages)
     extra = f"\n\nReviewer feedback I addressed:\n{feedback}" if feedback else ""
@@ -135,7 +163,6 @@ def _step_reply(messages: list[BaseMessage]) -> AIMessage:
     )
 
 
-# --- plan-mode flow --------------------------------------------------------
 PLAN_FILE_PATH = "/workspace/plans/2026-06-29-greet-helper.md"
 
 PLAN_MARKDOWN = """## Plan: Add greet() helper
@@ -155,14 +182,7 @@ Add a tiny greeting helper to the demo repo.
 """
 
 
-def _step_enter_plan(_messages: list[BaseMessage]) -> AIMessage:
-    return AIMessage(
-        content="This is worth planning first — entering plan mode.",
-        tool_calls=[{"name": "enter_plan_mode", "args": {}, "id": "call-enter-plan"}],
-    )
-
-
-def _step_plan_link(messages: list[BaseMessage]) -> AIMessage:
+def _plan_link_step(messages: list[BaseMessage]) -> AIMessage:
     url = _plan_url_from_messages(messages) or "(plan link unavailable)"
     return AIMessage(
         content="Sharing the plan-review link.",
@@ -178,7 +198,7 @@ def _step_plan_link(messages: list[BaseMessage]) -> AIMessage:
     )
 
 
-def _step_plan_research(_messages: list[BaseMessage]) -> AIMessage:
+def _plan_research_step(_messages: list[BaseMessage]) -> AIMessage:
     return AIMessage(
         content="Reading the repo to ground the plan.",
         tool_calls=[
@@ -187,7 +207,7 @@ def _step_plan_research(_messages: list[BaseMessage]) -> AIMessage:
     )
 
 
-def _step_write_plan(_messages: list[BaseMessage]) -> AIMessage:
+def _write_plan_step(_messages: list[BaseMessage]) -> AIMessage:
     return AIMessage(
         content="Writing the plan file for review.",
         tool_calls=[
@@ -200,7 +220,7 @@ def _step_write_plan(_messages: list[BaseMessage]) -> AIMessage:
     )
 
 
-def _step_save_plan(_messages: list[BaseMessage]) -> AIMessage:
+def _save_plan_step(_messages: list[BaseMessage]) -> AIMessage:
     return AIMessage(
         content="Saving the plan for review.",
         tool_calls=[
@@ -213,7 +233,7 @@ def _step_save_plan(_messages: list[BaseMessage]) -> AIMessage:
     )
 
 
-def _step_plan_complete(messages: list[BaseMessage]) -> AIMessage:
+def _plan_complete_step(messages: list[BaseMessage]) -> AIMessage:
     url = _plan_url_from_messages(messages) or "(plan link unavailable)"
     return AIMessage(
         content="Announcing the plan is ready.",
@@ -230,24 +250,7 @@ def _step_plan_complete(messages: list[BaseMessage]) -> AIMessage:
     )
 
 
-def _step_plan_end(_messages: list[BaseMessage]) -> AIMessage:
-    return AIMessage(content="I'll wait for your review and approval before implementing.")
-
-
-def build_plan_script() -> list[Any]:
-    return [
-        _step_enter_plan,
-        _step_plan_link,
-        _step_plan_research,
-        _step_write_plan,
-        _step_save_plan,
-        _step_plan_complete,
-        _step_plan_end,
-    ]
-
-
 FOLLOW_UP_REPLY = "Thanks! The PR is ready for review — anything else you'd like changed?"
-_ATTRIBUTION_RE = re.compile(r"@([A-Za-z0-9-]+):")
 
 
 def _latest_attribution(messages: list[BaseMessage]) -> str | None:
@@ -259,7 +262,7 @@ def _latest_attribution(messages: list[BaseMessage]) -> str | None:
     return None
 
 
-def _step_followup(messages: list[BaseMessage]) -> AIMessage:
+def _followup_step(messages: list[BaseMessage]) -> AIMessage:
     if any(
         isinstance(msg, HumanMessage) and "Please queue this follow-up" in _text(msg.content)
         for msg in messages
@@ -270,12 +273,105 @@ def _step_followup(messages: list[BaseMessage]) -> AIMessage:
     return AIMessage(content=f"{FOLLOW_UP_REPLY}{suffix}")
 
 
-def build_script() -> list[Any]:
-    return [_step_implement, _step_open_pr, _step_reply]
+SCRIPT_LIBRARY: dict[str, tuple[StepSpec, ...]] = {
+    "implement": (
+        _tool_step(
+            "Setting up the repo and implementing the change.",
+            "execute",
+            {"command": _IMPLEMENT_SCRIPT},
+            "call-impl",
+        ),
+        _tool_step(
+            "Opening a pull request.",
+            "open_pull_request",
+            {
+                "owner": OWNER,
+                "repo": REPO,
+                "head": FEATURE_BRANCH,
+                "base": BASE_BRANCH,
+                "title": PR_TITLE,
+                "body": "Adds a `greet()` helper as requested.",
+                "draft": True,
+            },
+            "call-pr",
+        ),
+        _dynamic_step(_reply_step),
+    ),
+    "breakout": (
+        _tool_step(
+            "Starting a separate Slack thread for the breakout task.",
+            "slack_start_new_thread",
+            {
+                "title": "Add greet() helper",
+                "instructions": "Please add a greet() helper and open a draft PR in the default repository. Use the current Slack request as context, and report progress in this new thread.",
+            },
+            "call-breakout",
+        ),
+        _tool_step(
+            "Confirming the breakout thread was started.",
+            "slack_thread_reply",
+            {"message": "I started a separate Open SWE thread for that aspect."},
+            "call-breakout-reply",
+        ),
+    ),
+    "plan": (
+        _tool_step(
+            "This is worth planning first — entering plan mode.",
+            "enter_plan_mode",
+            {},
+            "call-enter-plan",
+        ),
+        _dynamic_step(_plan_link_step),
+        _dynamic_step(_plan_research_step),
+        _dynamic_step(_write_plan_step),
+        _dynamic_step(_save_plan_step),
+        _dynamic_step(_plan_complete_step),
+        StepSpec(content="I'll wait for your review and approval before implementing."),
+    ),
+    "followup": (_dynamic_step(_followup_step),),
+}
 
 
-def build_followup_script() -> list[Any]:
-    return [_step_followup]
+def _is_plan_request(text: str) -> bool:
+    return "plan" in text.lower()
+
+
+def _is_breakout_request(text: str) -> bool:
+    t = text.lower()
+    return "break out" in t or "separate thread" in t or "split out" in t
+
+
+def _is_approval(text: str) -> bool:
+    t = text.lower()
+    return "approved" in t and "implement" in t
+
+
+def _is_revision(text: str) -> bool:
+    t = text.lower()
+    return "needs changes" in t or "publish an updated plan" in t
+
+
+SCRIPT_RULES: tuple[ScriptRule, ...] = (
+    ScriptRule("implement", lambda ctx: _is_approval(ctx.last_text)),
+    ScriptRule("plan", lambda ctx: _is_revision(ctx.last_text)),
+    ScriptRule("plan", lambda ctx: ctx.human_count <= 1 and _is_plan_request(ctx.first_text)),
+    ScriptRule(
+        "breakout", lambda ctx: ctx.human_count <= 1 and _is_breakout_request(ctx.first_text)
+    ),
+    ScriptRule("implement", lambda ctx: ctx.human_count <= 1),
+    ScriptRule("followup", lambda _ctx: True),
+)
+
+
+def _script_for(context: ScriptContext) -> tuple[StepSpec, ...]:
+    for rule in SCRIPT_RULES:
+        if rule.predicate(context):
+            return SCRIPT_LIBRARY[rule.name]
+    return SCRIPT_LIBRARY["followup"]
+
+
+def build_script() -> list[StepSpec]:
+    return list(SCRIPT_LIBRARY["implement"])
 
 
 class FakeScriptedChatModel(BaseChatModel):
@@ -298,42 +394,16 @@ class FakeScriptedChatModel(BaseChatModel):
         **kwargs: Any,
     ) -> ChatResult:
         humans = [m for m in messages if isinstance(m, HumanMessage)]
-        first_text = _text(humans[0].content) if humans else ""
-        last_text = _text(humans[-1].content) if humans else ""
+        context = ScriptContext(
+            first_text=_text(humans[0].content) if humans else "",
+            last_text=_text(humans[-1].content) if humans else "",
+            human_count=len(humans),
+        )
+        script = _script_for(context)
 
-        # Pick the script for the current turn by what the latest human asked.
-        if _is_approval(last_text):
-            script = build_script()  # implement + open PR + reply
-        elif _is_revision(last_text):
-            script = build_plan_script()  # re-plan after requested changes
-        elif _is_plan_request(first_text) and len(humans) <= 1:
-            script = build_plan_script()  # first ask was to plan
-        elif len(humans) <= 1:
-            script = build_script()
-        else:
-            script = build_followup_script()
-
-        # Step within the *current* turn: AIMessages since the last human turn.
         last_human = max(
             (i for i, m in enumerate(messages) if isinstance(m, HumanMessage)), default=-1
         )
-        step = sum(1 for m in messages[last_human + 1 :] if isinstance(m, AIMessage))
-        if step < len(script):
-            message = script[step](messages)
-        else:
-            message = _step_followup(messages)
-        return ChatResult(generations=[ChatGeneration(message=message)])
-
-
-def _is_plan_request(text: str) -> bool:
-    return "plan" in text.lower()
-
-
-def _is_approval(text: str) -> bool:
-    t = text.lower()
-    return "approved" in t and "implement" in t
-
-
-def _is_revision(text: str) -> bool:
-    t = text.lower()
-    return "needs changes" in t or "publish an updated plan" in t
+        step_index = sum(1 for m in messages[last_human + 1 :] if isinstance(m, AIMessage))
+        step = script[step_index] if step_index < len(script) else SCRIPT_LIBRARY["followup"][0]
+        return ChatResult(generations=[ChatGeneration(message=_render_step(step, messages))])
diff --git a/tests/e2e/fakes.py b/tests/e2e/fakes.py
index d1e1a85b..b1dde372 100644
--- a/tests/e2e/fakes.py
+++ b/tests/e2e/fakes.py
@@ -41,12 +41,13 @@ def add_slack_message(
     channel: str, thread_ts: str, *, user: str, text: str, blocks: Any = None, is_bot: bool = False
 ) -> str:
     ts = next_slack_ts()
-    SLACK_MESSAGES.setdefault((channel, thread_ts), []).append(
+    actual_thread_ts = thread_ts or ts
+    SLACK_MESSAGES.setdefault((channel, actual_thread_ts), []).append(
         {
             "user": user,
             "text": text,
             "ts": ts,
-            "thread_ts": thread_ts,
+            "thread_ts": actual_thread_ts,
             "blocks": blocks,
             "is_bot": is_bot,
         }
@@ -58,6 +59,14 @@ def slack_thread(channel: str, thread_ts: str) -> list[dict[str, Any]]:
     return SLACK_MESSAGES.get((channel, thread_ts), [])
 
 
+def slack_messages(channel: str) -> list[dict[str, Any]]:
+    messages: list[dict[str, Any]] = []
+    for (message_channel, _thread_ts), thread_messages in SLACK_MESSAGES.items():
+        if message_channel == channel:
+            messages.extend(thread_messages)
+    return sorted(messages, key=lambda message: message["ts"])
+
+
 # --- GitHub ----------------------------------------------------------------
 PULLS: list[dict[str, Any]] = []
 _pr_seq = [0]
@@ -149,5 +158,4 @@ def reset() -> None:
     SLACK_MESSAGES.clear()
     PULLS.clear()
     _pr_seq[0] = 0
-    _slack_seq[0] = 1
     seed_bare_remote()
diff --git a/tests/e2e/harness.py b/tests/e2e/harness.py
index 2730d0fa..1147300b 100644
--- a/tests/e2e/harness.py
+++ b/tests/e2e/harness.py
@@ -309,10 +309,18 @@ async def mock_users() -> JSONResponse:
 
 @app.get("/mock/slack/messages")
 async def slack_messages() -> JSONResponse:
-    thread = CURRENT_THREAD["thread_ts"]
-    msgs = fakes.slack_thread(CURRENT_THREAD["channel"], thread) if thread else []
+    msgs = fakes.slack_messages(CURRENT_THREAD["channel"])
     return JSONResponse(
-        [{"user": m["user"], "text": m["text"], "is_bot": m["is_bot"], "ts": m["ts"]} for m in msgs]
+        [
+            {
+                "user": m["user"],
+                "text": m["text"],
+                "is_bot": m["is_bot"],
+                "ts": m["ts"],
+                "thread_ts": m["thread_ts"],
+            }
+            for m in msgs
+        ]
     )
 
 
diff --git a/tests/e2e/static/slack.html b/tests/e2e/static/slack.html
index ba806672..34093cba 100644
--- a/tests/e2e/static/slack.html
+++ b/tests/e2e/static/slack.html
@@ -47,7 +47,7 @@
           .map((m) => {
             const who = m.is_bot ? "open-swe (bot)" : userNames[m.user] || m.user;
             const linked = m.text.replace(/<(https?:\/\/[^|>]+)\|([^>]+)>/g, '<a href="$1">$2</a>');
-            return `<div class="msg ${m.is_bot ? "bot" : ""}" data-bot="${m.is_bot}"><div class="who">${who}</div><div class="text">${linked}</div></div>`;
+            return `<div class="msg ${m.is_bot ? "bot" : ""}" data-bot="${m.is_bot}" data-thread-ts="${m.thread_ts || ""}"><div class="who">${who}</div><div class="text">${linked}</div></div>`;
           })
           .join("");
         $("thread").innerHTML = html || "<p style='color:#888'>No messages yet.</p>";
diff --git a/tests/e2e/tests/full_flow.spec.ts b/tests/e2e/tests/full_flow.spec.ts
index f3602eaf..959e6aeb 100644
--- a/tests/e2e/tests/full_flow.spec.ts
+++ b/tests/e2e/tests/full_flow.spec.ts
@@ -37,6 +37,26 @@ test.describe("Open SWE full flow", () => {
     await expect(page.locator('.pr[data-pr="1"]')).toContainText("greet.py");
   });
 
+  test("Slack breakout request starts a new top-level Open SWE thread", async ({ page }) => {
+    await page.locator("#text").fill("<@U0BOT> please break out adding a greet() helper into a separate thread");
+    await page.locator("#send").click();
+
+    const breakout = page
+      .locator(".msg.bot")
+      .filter({ hasText: /Open SWE breakout thread:\* Add greet\(\) helper/ });
+    await expect(breakout).toBeVisible({ timeout: 60_000 });
+    const breakoutThreadTs = await breakout.getAttribute("data-thread-ts");
+    expect(breakoutThreadTs).toBeTruthy();
+
+    const breakoutThreadMessages = page.locator(`.msg.bot[data-thread-ts="${breakoutThreadTs}"]`);
+    await expect(breakoutThreadMessages.locator('a[href*="/agents/"]')).toBeVisible({
+      timeout: 60_000,
+    });
+    await expect(
+      page.locator(".msg.bot").filter({ hasText: "I started a separate Open SWE thread" }),
+    ).toBeVisible({ timeout: 60_000 });
+  });
+
   test("a message that does not mention the bot produces no run and no PR", async ({ page }) => {
     await page.locator("#mention").uncheck();
     await page.locator("#text").fill("just chatting with the team, nothing for the bot");
diff --git a/tests/test_plan_mode.py b/tests/test_plan_mode.py
index 8a6e2d68..a5c6a4ba 100644
--- a/tests/test_plan_mode.py
+++ b/tests/test_plan_mode.py
@@ -27,6 +27,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_assistants_status.py b/tests/test_slack_assistants_status.py
index 4bdea0e1..4d3c592e 100644
--- a/tests/test_slack_assistants_status.py
+++ b/tests/test_slack_assistants_status.py
@@ -271,6 +271,36 @@ async def test_post_slack_thread_reply_with_ts_sends_blocks(
     assert client_cm.post.call_args.kwargs["json"]["blocks"] == blocks
 
 
+@pytest.mark.asyncio
+async def test_post_slack_top_level_message_with_ts_omits_thread_ts(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setattr(slack_utils, "SLACK_BOT_TOKEN", "xoxb-test")
+
+    client_cm = _async_client_cm(_ok_response())
+    with patch.object(slack_utils.httpx, "AsyncClient", return_value=client_cm):
+        result = await slack_utils.post_slack_top_level_message_with_ts("C1", "hello")
+
+    assert result == ("1.0", None)
+    payload = client_cm.post.call_args.kwargs["json"]
+    assert payload["channel"] == "C1"
+    assert payload["text"] == "hello"
+    assert "thread_ts" not in payload
+
+
+@pytest.mark.asyncio
+async def test_post_slack_top_level_message_with_ts_returns_slack_error(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setattr(slack_utils, "SLACK_BOT_TOKEN", "xoxb-test")
+
+    client_cm = _async_client_cm(_err_response("msg_too_long"))
+    with patch.object(slack_utils.httpx, "AsyncClient", return_value=client_cm):
+        result = await slack_utils.post_slack_top_level_message_with_ts("C1", "hello")
+
+    assert result == (None, "msg_too_long")
+
+
 async def test_post_slack_thread_reply_preserves_bool_return_on_error(
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:
diff --git a/tests/test_slack_context.py b/tests/test_slack_context.py
index 511d04c7..da21ee10 100644
--- a/tests/test_slack_context.py
+++ b/tests/test_slack_context.py
@@ -17,9 +17,6 @@ from agent.utils.slack import (
 )
 from agent.webapp import generate_thread_id_from_slack_thread
 
-_TEXT_ONLY_MODEL = "fireworks:accounts/fireworks/models/glm-5p2"
-_VISION_MODEL = "openai:gpt-5.5"
-
 
 class _FakeNotFoundError(Exception):
     status_code = 404
@@ -45,22 +42,6 @@ class _FakeClient:
         self.threads = threads_client
 
 
-class _FakeSlackMappingStore:
-    def __init__(self) -> None:
-        self.items: dict[tuple[tuple[str, ...], str], dict] = {}
-
-    async def put_item(self, namespace: tuple[str, ...], key: str, value: dict) -> None:
-        self.items[(namespace, key)] = {"value": value}
-
-    async def get_item(self, namespace: tuple[str, ...], key: str) -> dict | None:
-        return self.items.get((namespace, key))
-
-
-class _FakeSlackMappingClient:
-    def __init__(self) -> None:
-        self.store = _FakeSlackMappingStore()
-
-
 def test_generate_thread_id_from_slack_thread_is_deterministic() -> None:
     channel_id = "C12345"
     thread_ts = "1730900000.123456"
@@ -70,63 +51,6 @@ def test_generate_thread_id_from_slack_thread_is_deterministic() -> None:
     assert len(first) == 36
 
 
-@pytest.mark.asyncio
-async def test_slack_run_mapping_preserves_trace_message_ts() -> None:
-    client = _FakeSlackMappingClient()
-
-    await slack_utils.store_slack_run_mapping(
-        client,
-        "C123",
-        "1.0",
-        "run-1",
-        message_ts="1.1",
-        triggering_user_id="U123",
-        trace_message_ts="1.1",
-    )
-    await slack_utils.store_slack_message_run_mapping(client, "C123", "1.0", "1.2")
-
-    thread_mapping = await slack_utils.lookup_slack_thread_run_mapping(client, "C123", "1.0")
-    message_mapping = await slack_utils.lookup_slack_run_mapping(client, "C123", "1.2")
-    assert thread_mapping is not None
-    assert thread_mapping["trace_message_ts"] == "1.1"
-    assert thread_mapping["triggering_user_id"] == "U123"
-    assert message_mapping is not None
-    assert message_mapping["trace_message_ts"] == "1.1"
-    assert message_mapping["message_ts"] == "1.2"
-
-
-@pytest.mark.asyncio
-async def test_slack_run_mapping_preserves_trace_message_ts_on_followup_mention() -> None:
-    """A subsequent Slack mention without trace_message_ts must not clobber the stored timestamp."""
-    client = _FakeSlackMappingClient()
-
-    # First mention stores the trace message ts.
-    await slack_utils.store_slack_run_mapping(
-        client,
-        "C123",
-        "1.0",
-        "run-1",
-        message_ts="1.1",
-        triggering_user_id="U123",
-        trace_message_ts="1.1",
-    )
-
-    # Follow-up mention (non-first) stores a new run_id without trace_message_ts.
-    await slack_utils.store_slack_run_mapping(
-        client,
-        "C123",
-        "1.0",
-        "run-2",
-        triggering_user_id="U456",
-    )
-
-    thread_mapping = await slack_utils.lookup_slack_thread_run_mapping(client, "C123", "1.0")
-    assert thread_mapping is not None
-    assert thread_mapping["run_id"] == "run-2"
-    assert thread_mapping["trace_message_ts"] == "1.1"
-    assert thread_mapping["triggering_user_id"] == "U456"
-
-
 def test_select_slack_context_messages_uses_thread_start_when_no_prior_mention() -> None:
     bot_user_id = "UBOT"
     messages = [
@@ -321,60 +245,6 @@ def test_post_slack_trace_reply_includes_trace_link_and_tip(
     assert posted[0]["unfurl_media"] is False
 
 
-def test_format_trace_reply_can_show_web_handoff_notice() -> None:
-    text = slack_utils._format_trace_reply(
-        "https://smith/x", "https://app.example.com/agents/thread-id", moved_to_web=True
-    )
-
-    head, _, notice_line = text.partition("\n")
-    assert (
-        head
-        == "<https://smith/x|View trace> • <https://app.example.com/agents/thread-id|Open in Web>"
-    )
-    assert "Conversation moved to Web" in notice_line
-    assert "Tip:" not in notice_line
-
-
-@pytest.mark.asyncio
-async def test_update_slack_trace_reply_for_web_handoff_updates_existing_message(
-    monkeypatch: pytest.MonkeyPatch,
-) -> None:
-    updated: list[dict] = []
-
-    async def fake_update_slack_message(
-        channel_id: str,
-        message_ts: str,
-        text: str,
-        *,
-        unfurl_links: bool = True,
-        unfurl_media: bool = True,
-    ) -> tuple[bool, str | None]:
-        updated.append(
-            {
-                "channel_id": channel_id,
-                "message_ts": message_ts,
-                "text": text,
-                "unfurl_links": unfurl_links,
-                "unfurl_media": unfurl_media,
-            }
-        )
-        return True, None
-
-    monkeypatch.setenv("DASHBOARD_BASE_URL", "https://app.example.com")
-    monkeypatch.setattr(slack_utils, "update_slack_message", fake_update_slack_message)
-    monkeypatch.setattr(slack_utils, "get_langsmith_trace_url", lambda thread_id: "https://smith/x")
-
-    ok = await slack_utils.update_slack_trace_reply_for_web_handoff("C123", "1.1", "thread-id")
-
-    assert ok is True
-    assert len(updated) == 1
-    assert updated[0]["channel_id"] == "C123"
-    assert updated[0]["message_ts"] == "1.1"
-    assert "Conversation moved to Web" in updated[0]["text"]
-    assert updated[0]["unfurl_links"] is False
-    assert updated[0]["unfurl_media"] is False
-
-
 def test_post_slack_trace_reply_can_skip_web_link(
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:
@@ -505,35 +375,6 @@ def test_get_slack_repo_config_ignores_repo_syntax_in_message(
     assert repo == {"owner": "saved-owner", "name": "saved-repo"}
 
 
-def test_get_slack_repo_config_uses_prefetched_channel_context(
-    monkeypatch: pytest.MonkeyPatch,
-) -> None:
-    threads_client = _FakeThreadsClient(thread={"metadata": {}})
-
-    async def fail_get_slack_channel_description(channel_id: str) -> str:
-        raise AssertionError("prefetched channel context should avoid a duplicate Slack lookup")
-
-    monkeypatch.setattr(webapp, "get_client", lambda url: _FakeClient(threads_client))
-    monkeypatch.setattr(webapp, "get_slack_channel_description", fail_get_slack_channel_description)
-
-    repo = asyncio.run(
-        webapp.get_slack_repo_config(
-            "C123",
-            "1.234",
-            channel_context={
-                "id": "C123",
-                "name": "eng-open-swe",
-                "name_normalized": "eng-open-swe",
-                "topic": "repo:langchain-ai/open-swe",
-                "purpose": "agent work",
-                "description": "repo:langchain-ai/open-swe\nagent work",
-            },
-        )
-    )
-
-    assert repo == {"owner": "langchain-ai", "name": "open-swe"}
-
-
 def test_get_slack_repo_config_applies_profile_default_repo(
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:
@@ -689,14 +530,6 @@ def test_process_slack_mention_creates_thread_first_run_with_trace_reply(
         webapp.process_slack_mention(
             {
                 "channel_id": "C123",
-                "channel_context": {
-                    "id": "C123",
-                    "name": "Eng Open SWE",
-                    "name_normalized": "eng-open-swe",
-                    "topic": "Coordinate Open SWE work",
-                    "purpose": "repo:langchain-ai/open-swe",
-                    "description": "Coordinate Open SWE work\nrepo:langchain-ai/open-swe",
-                },
                 "thread_ts": thread_ts,
                 "event_ts": event_ts,
                 "user_id": "U123",
@@ -727,9 +560,7 @@ def test_process_slack_mention_creates_thread_first_run_with_trace_reply(
     assert kwargs["if_not_exists"] == "create"
     assert kwargs["multitask_strategy"] == "interrupt"
     assert kwargs["durability"] == "sync"
-    slack_thread_config = kwargs["config"]["configurable"]["slack_thread"]
-    assert slack_thread_config["thread_ts"] == thread_ts
-    assert slack_thread_config["channel_context"]["name_normalized"] == "eng-open-swe"
+    assert kwargs["config"]["configurable"]["slack_thread"]["thread_ts"] == thread_ts
     prompt_block = kwargs["input"]["messages"][0]["content"][0]
     assert "## Default Repository Hint\nlangchain-ai/open-swe" in prompt_block["text"]
     assert (
@@ -737,47 +568,10 @@ def test_process_slack_mention_creates_thread_first_run_with_trace_reply(
         in (prompt_block["text"])
     )
     assert prompt_block["text"].count("## Slack Thread") == 1
-    assert "Channel ID: C123" in prompt_block["text"]
-    assert "Channel name: #eng-open-swe" in prompt_block["text"]
     assert f"Thread TS: {thread_ts}" in prompt_block["text"]
-    assert "Slack-provided channel description" in prompt_block["text"]
-    assert "Coordinate Open SWE work" in prompt_block["text"]
-    assert "repo:langchain-ai/open-swe" in prompt_block["text"]
     assert "## Latest Mention Request\ncontinue on the branch" in prompt_block["text"]
 
 
-def test_process_slack_mention_prompt_omits_missing_channel_metadata(
-    monkeypatch: pytest.MonkeyPatch,
-) -> None:
-    captured: dict[str, object] = {}
-    _setup_slack_mention_fakes(monkeypatch, captured)
-
-    async def fake_thread_exists(thread_id: str) -> bool:
-        return False
-
-    monkeypatch.setattr(webapp, "_thread_exists", fake_thread_exists)
-
-    asyncio.run(
-        webapp.process_slack_mention(
-            {
-                "channel_id": "C123",
-                "thread_ts": "1700000000.000100",
-                "event_ts": "1700000000.000200",
-                "user_id": "U123",
-                "text": "<@UBOT> do the thing",
-                "bot_user_id": "UBOT",
-            },
-            {"owner": "langchain-ai", "name": "open-swe"},
-        )
-    )
-
-    run_create = captured["run_create"]
-    prompt_block = run_create["kwargs"]["input"]["messages"][0]["content"][0]
-    assert "Channel ID: C123" in prompt_block["text"]
-    assert "Channel name:" not in prompt_block["text"]
-    assert "Slack-provided channel description" not in prompt_block["text"]
-
-
 def test_process_slack_mention_skips_trace_reply_on_followup_mention(
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:
@@ -979,20 +773,10 @@ def test_process_slack_mention_mapped_user_with_token_runs_as_user(
     monkeypatch.setattr(webapp, "login_for_slack_id", fake_login_for_slack_id)
     monkeypatch.setattr(webapp, "upsert_agent_thread_owner_metadata", fake_upsert_owner)
 
-    channel_context = {
-        "id": "C123",
-        "name": "eng-open-swe",
-        "name_normalized": "eng-open-swe",
-        "topic": "Coordinate Open SWE work",
-        "purpose": "",
-        "description": "Coordinate Open SWE work",
-    }
-
     asyncio.run(
         webapp.process_slack_mention(
             {
                 "channel_id": "C123",
-                "channel_context": channel_context,
                 "thread_ts": "1700000000.000100",
                 "event_ts": "1700000000.000200",
                 "user_id": "U123",
@@ -1006,10 +790,6 @@ def test_process_slack_mention_mapped_user_with_token_runs_as_user(
     run_create = captured["run_create"]
     configurable = run_create["kwargs"]["config"]["configurable"]
     assert configurable["github_login"] == "mason-gh"
-    assert configurable["slack_thread"]["channel_context"] == channel_context
-    assert owner_meta["source_context"] == {
-        "slack_thread": configurable["slack_thread"],
-    }
     # The thread is tagged with the login resolved from the Slack user id, so it
     # surfaces in the web Agents UI even when the Slack profile email does not
     # resolve to a mapping (login_for_email returns None in this harness).
@@ -1057,76 +837,9 @@ def test_process_slack_mention_bot_only_mode_runs_without_user_token(
     assert "prompt" not in captured
 
 
-def test_process_slack_mention_uses_vision_fallback_for_image_thread(
-    monkeypatch: pytest.MonkeyPatch,
-) -> None:
-    captured: dict[str, object] = {}
-    _setup_slack_mention_fakes(monkeypatch, captured)
-
-    async def fake_thread_exists(thread_id: str) -> bool:
-        return False
-
-    async def fake_fetch_slack_thread_messages(channel_id: str, thread_ts: str) -> list[dict]:
-        return [
-            {
-                "ts": "1700000000.000100",
-                "text": "<@UBOT> please inspect this",
-                "user": "U123",
-                "files": [
-                    {
-                        "mimetype": "image/png",
-                        "url_private": "https://files.slack.com/screenshot.png",
-                    }
-                ],
-            }
-        ]
-
-    async def fake_resolve_agent_model_id(login: str | None) -> str:
-        assert login == "mason-gh"
-        return _TEXT_ONLY_MODEL
-
-    async def fake_fetch_image_block(image_url: str, client: object) -> dict[str, str]:
-        captured["image_url"] = image_url
-        return {"type": "image", "source_type": "base64", "mime_type": "image/png", "data": "abc"}
-
-    monkeypatch.setattr(webapp, "_thread_exists", fake_thread_exists)
-    monkeypatch.setattr(webapp, "fetch_slack_thread_messages", fake_fetch_slack_thread_messages)
-    monkeypatch.setattr(webapp, "resolve_agent_model_id", fake_resolve_agent_model_id)
-    monkeypatch.setattr(webapp, "fetch_image_block", fake_fetch_image_block)
-
-    asyncio.run(
-        webapp.process_slack_mention(
-            {
-                "channel_id": "C123",
-                "thread_ts": "1700000000.000100",
-                "event_ts": "1700000000.000100",
-                "user_id": "U123",
-                "text": "<@UBOT> please inspect this",
-                "bot_user_id": "UBOT",
-            },
-            {"owner": "langchain-ai", "name": "open-swe"},
-        )
-    )
-
-    assert captured["image_url"] == "https://files.slack.com/screenshot.png"
-    run_create = captured["run_create"]
-    assert isinstance(run_create, dict)
-    kwargs = run_create["kwargs"]
-    configurable = kwargs["config"]["configurable"]
-    assert configurable["agent_model_id"] == _VISION_MODEL
-    assert configurable["agent_effort"] == "medium"
-    content = kwargs["input"]["messages"][0]["content"]
-    assert any(block.get("type") == "image" for block in content)
-    assert "does not support image input" not in content[0]["text"]
-
-
 class _FakeResponse:
-    def __init__(
-        self, payload: dict, status_code: int = 200, headers: dict[str, str] | None = None
-    ) -> None:
+    def __init__(self, payload: dict) -> None:
         self._payload = payload
-        self.status_code = status_code
-        self.headers = headers or {}
 
     def raise_for_status(self) -> None:
         return None
@@ -1149,67 +862,6 @@ class _FakeAsyncClient:
         return _FakeResponse(self._payload)
 
 
-def test_get_slack_channel_info_uses_global_ttl_cache(monkeypatch: pytest.MonkeyPatch) -> None:
-    slack_utils.clear_slack_channel_info_cache()
-    calls = 0
-    payload = {
-        "ok": True,
-        "channel": {
-            "id": "C123",
-            "name": "eng-open-swe",
-            "topic": {"value": "Coordinate work"},
-            "purpose": {"value": "repo:langchain-ai/open-swe"},
-        },
-    }
-
-    class _CountingAsyncClient:
-        async def __aenter__(self) -> "_CountingAsyncClient":
-            return self
-
-        async def __aexit__(self, *exc: object) -> None:
-            return None
-
-        async def get(self, url: str, **kwargs: object) -> _FakeResponse:
-            nonlocal calls
-            calls += 1
-            return _FakeResponse(payload)
-
-    monkeypatch.setattr(slack_utils, "SLACK_BOT_TOKEN", "xoxb-test")
-    monkeypatch.setattr(slack_utils.httpx, "AsyncClient", lambda *a, **k: _CountingAsyncClient())
-
-    first = asyncio.run(slack_utils.get_slack_channel_info("C123"))
-    second = asyncio.run(slack_utils.get_slack_channel_info("C123"))
-
-    assert first == payload["channel"]
-    assert second == payload["channel"]
-    assert calls == 1
-    slack_utils.clear_slack_channel_info_cache()
-
-
-def test_get_slack_channel_info_rate_limit_is_non_fatal(monkeypatch: pytest.MonkeyPatch) -> None:
-    slack_utils.clear_slack_channel_info_cache()
-
-    class _RateLimitedAsyncClient:
-        async def __aenter__(self) -> "_RateLimitedAsyncClient":
-            return self
-
-        async def __aexit__(self, *exc: object) -> None:
-            return None
-
-        async def get(self, url: str, **kwargs: object) -> _FakeResponse:
-            return _FakeResponse(
-                {"ok": False, "error": "ratelimited"},
-                status_code=429,
-                headers={"Retry-After": "30"},
-            )
-
-    monkeypatch.setattr(slack_utils, "SLACK_BOT_TOKEN", "xoxb-test")
-    monkeypatch.setattr(slack_utils.httpx, "AsyncClient", lambda *a, **k: _RateLimitedAsyncClient())
-
-    assert asyncio.run(slack_utils.get_slack_channel_info("C123")) is None
-    assert slack_utils._SLACK_CHANNEL_INFO_CACHE == {}
-
-
 def test_get_slack_permalink_returns_link(monkeypatch: pytest.MonkeyPatch) -> None:
     monkeypatch.setattr(slack_utils, "SLACK_BOT_TOKEN", "xoxb-test")
     link = "https://workspace.slack.com/archives/C123/p1700000000000100"
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..9f7c1c2d
--- /dev/null
+++ b/tests/test_slack_start_new_thread_tool.py
@@ -0,0 +1,256 @@
+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": "anthropic:claude-sonnet-4-5",
+            "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"] == "anthropic:claude-sonnet-4-5"
+    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