Merge remote-tracking branch 'origin/dev' into feature/80-modular-webhook-architecture

A burst of near-simultaneous CI events for one head SHA can slip past
the busy-check before the dedupe SHA is recorded, so dispatch the
autofix path with multitask_strategy=reject (dev's prior platform
default) to drop duplicate concurrent creates instead of letting them
interrupt each other. Also make the completion failure-reply dedup
claim-then-post and drop the unreachable interrupted branch.

agent/ci_autofix.py

@@ -280,11 +280,17 @@ async def _dispatch_or_batch(
         logger.info("Agent thread %s busy; batching auto-fix event %s", thread_id, reason)
         await _mark_pending_autofix_event(thread_id, reason, detail)
         return "batched"
+    # The busy-check above has a TOCTOU window (the dedupe SHA is only recorded
+    # after dispatch), so a burst of near-simultaneous CI events for one head SHA
+    # can all pass the gate. Dispatch with ``reject`` — matching ``dev``'s prior
+    # platform default — so the platform drops the duplicate concurrent creates
+    # instead of letting them interrupt each other.
     await dispatch_agent_run(
         thread_id,
         prompt,
         configurable,
         source=str(configurable.get("source") or "github_autofix"),
+        multitask_strategy="reject",
     )
     logger.info(
         "Created auto-fix run for thread %s (source=%s)", thread_id, configurable.get("source")

agent/completion.py

@@ -16,6 +16,7 @@ from __future__ import annotations
 import hmac
 import logging
 import os
+from collections.abc import Awaitable, Callable
 from typing import Any
 
 from .utils.github_app import get_github_app_installation_token
@@ -33,6 +34,11 @@ logger = logging.getLogger(__name__)
 _TERMINAL_FAILURE_STATUSES = frozenset({"error", "timeout"})
 _FAILURE_REPLY_FLAG = "failure_reply_posted"
 
+
+class _ClaimFailed(Exception):
+    """Raised when the dedup flag couldn't be claimed, so we skip the post."""
+
+
 # Shared-secret bearer token proving a /webhooks/run-complete call came from our
 # own dispatch (which appends ?token= when this is set) rather than from an
 # attacker hitting the public route. Fail closed when unset: the route rejects
@@ -58,20 +64,26 @@ def verify_run_complete_token(token: str | None) -> bool:
 
 
 def _failure_text(status: str) -> str:
-    if status == "timeout":
-        reason = "timed out"
-    elif status == "interrupted":
-        reason = "was interrupted before it could finish"
-    else:
-        reason = "hit an unexpected error"
+    reason = "timed out" if status == "timeout" else "hit an unexpected error"
     return (
         f"⚠️ I wasn't able to finish that — the run {reason}. "
         "Send another message and I'll pick it back up."
     )
 
 
-async def _post_failure_reply(thread_id: str, metadata: dict[str, Any], status: str) -> bool:
-    """Post a failure reply to the run's originating channel. Best-effort."""
+async def _post_failure_reply(
+    thread_id: str,
+    metadata: dict[str, Any],
+    status: str,
+    *,
+    claim: Callable[[], Awaitable[None]],
+) -> bool:
+    """Post a failure reply to the run's originating channel. Best-effort.
+
+    ``claim`` is awaited immediately before the network post (claim-then-post),
+    only on a branch that actually delivers, so a retried/concurrent webhook
+    can't double-post and threads with no channel never burn the flag.
+    """
     source = metadata.get("source")
     ctx = metadata.get("source_context")
     ctx = ctx if isinstance(ctx, dict) else {}
@@ -83,6 +95,7 @@ async def _post_failure_reply(thread_id: str, metadata: dict[str, Any], status:
             channel_id = slack_thread.get("channel_id")
             thread_ts = slack_thread.get("thread_ts")
             if channel_id and thread_ts:
+                await claim()
                 return await post_slack_thread_reply(channel_id, thread_ts, text)
         return False
 
@@ -91,6 +104,7 @@ async def _post_failure_reply(thread_id: str, metadata: dict[str, Any], status:
         if isinstance(linear_issue, dict):
             issue_id = linear_issue.get("id")
             if issue_id:
+                await claim()
                 return await comment_on_linear_issue(issue_id, text)
         return False
 
@@ -104,6 +118,7 @@ async def _post_failure_reply(thread_id: str, metadata: dict[str, Any], status:
         if isinstance(repo_config, dict) and isinstance(number, int):
             token = await get_github_app_installation_token()
             if token:
+                await claim()
                 return await post_github_comment(repo_config, number, text, token=token)
         return False
 
@@ -136,13 +151,23 @@ async def handle_run_completion(payload: dict[str, Any]) -> dict[str, str]:
     if metadata.get(_FAILURE_REPLY_FLAG):
         return {"status": "ignored", "reason": "failure reply already posted"}
 
-    posted = await _post_failure_reply(thread_id, metadata, status)
-    if not posted:
-        return {"status": "ignored", "reason": "no reply posted"}
+    # Claim-then-post: set the dedup flag immediately before the actual post (via
+    # the claim callback) so a retried/concurrent completion webhook can't
+    # double-post. The flag is only claimed on a branch that delivers, so a
+    # thread with no reply channel never burns it. If the claim itself fails we
+    # skip the post, leaving the flag unset so a later retry can try again.
+    async def _claim() -> None:
+        try:
+            await client.threads.update(thread_id=thread_id, metadata={_FAILURE_REPLY_FLAG: True})
+        except Exception as exc:  # noqa: BLE001
+            logger.warning("run-complete: could not flag thread %s", thread_id, exc_info=True)
+            raise _ClaimFailed from exc
 
     try:
-        await client.threads.update(thread_id=thread_id, metadata={_FAILURE_REPLY_FLAG: True})
-    except Exception:  # noqa: BLE001
-        logger.warning("run-complete: could not flag thread %s", thread_id, exc_info=True)
+        posted = await _post_failure_reply(thread_id, metadata, status, claim=_claim)
+    except _ClaimFailed:
+        return {"status": "error", "reason": "could not claim failure reply"}
+    if not posted:
+        return {"status": "ignored", "reason": "no reply posted"}
     logger.info("Posted failure reply for thread %s (status=%s)", thread_id, status)
     return {"status": "ok", "reason": "failure reply posted"}

agent/dashboard/schedules.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import logging
+import re
 import uuid
 from datetime import UTC, datetime
 from typing import Any
@@ -22,6 +23,7 @@ SCHEDULES_NAMESPACE: list[str] = ["agent_schedules"]
 _AGENT_ASSISTANT_ID = "agent"
 _SCHEDULER_ASSISTANT_ID = "scheduler"
 _CRON_FIELD_RANGES = ((0, 59), (0, 23), (1, 31), (1, 12), (0, 7))
+_SLACK_CHANNEL_ID_RE = re.compile(r"^[A-Z][A-Z0-9]{5,}$")
 
 
 class ScheduleCreateBody(BaseModel):
@@ -31,12 +33,18 @@ class ScheduleCreateBody(BaseModel):
     repo: str | None = None
     model_id: str | None = None
     effort: str | None = None
+    slack_report_channel: str | None = Field(default=None, max_length=120)
 
     @field_validator("schedule")
     @classmethod
     def _valid_schedule(cls, value: str) -> str:
         return normalize_cron_schedule(value)
 
+    @field_validator("slack_report_channel")
+    @classmethod
+    def _valid_slack_report_channel(cls, value: str | None) -> str | None:
+        return normalize_slack_channel_id(value)
+
 
 class ScheduleUpdateBody(BaseModel):
     prompt: str | None = Field(default=None, min_length=1, max_length=20_000)
@@ -46,12 +54,18 @@ class ScheduleUpdateBody(BaseModel):
     model_id: str | None = None
     effort: str | None = None
     enabled: bool | None = None
+    slack_report_channel: str | None = Field(default=None, max_length=120)
 
     @field_validator("schedule")
     @classmethod
     def _valid_schedule(cls, value: str | None) -> str | None:
         return normalize_cron_schedule(value) if value is not None else None
 
+    @field_validator("slack_report_channel")
+    @classmethod
+    def _valid_slack_report_channel(cls, value: str | None) -> str | None:
+        return normalize_slack_channel_id(value)
+
 
 def _client():
     return langgraph_client()
@@ -109,6 +123,18 @@ def normalize_cron_schedule(raw: str) -> str:
     return value
 
 
+def normalize_slack_channel_id(raw: str | None) -> str | None:
+    """Normalize a Slack channel ID; blank becomes None (no report channel)."""
+    if raw is None:
+        return None
+    value = raw.strip().lstrip("#")
+    if not value:
+        return None
+    if not _SLACK_CHANNEL_ID_RE.match(value):
+        raise ValueError("slack_report_channel must be a Slack channel ID (e.g. C0123ABCD)")
+    return value
+
+
 def _derive_name(prompt: str) -> str:
     return prompt.strip().splitlines()[0][:80] or "Scheduled agent"
 
@@ -131,6 +157,7 @@ def _schedule_summary(record: dict[str, Any]) -> dict[str, Any]:
         "repo": _repo_full_name(repo),
         "model": record.get("model"),
         "effort": record.get("effort"),
+        "slackReportChannel": record.get("slack_report_channel"),
         "enabled": bool(record.get("enabled")),
         "cronId": record.get("cron_id"),
         "lastThreadId": record.get("last_thread_id"),
@@ -274,6 +301,7 @@ async def create_agent_schedule(
         "repo": repo,
         "model": chosen_model or profile.get("default_model") or "Default",
         "effort": chosen_effort or profile.get("reasoning_effort"),
+        "slack_report_channel": body.slack_report_channel,
         "base_branch": profile.get("base_branch") or "main",
         "branch_prefix": profile.get("branch_prefix"),
         "enabled": True,
@@ -322,6 +350,8 @@ async def update_agent_schedule(
             patch["effort"] = effort
     if body.enabled is not None:
         patch["enabled"] = body.enabled
+    if "slack_report_channel" in body.model_fields_set:
+        patch["slack_report_channel"] = body.slack_report_channel
 
     updated = {**existing, **patch}
     schedule_changed = updated.get("schedule") != existing.get("schedule")
@@ -390,6 +420,9 @@ def _agent_run_config(record: dict[str, Any], thread_id: str) -> dict[str, Any]:
     if model and effort:
         configurable["agent_model_id"] = model
         configurable["agent_effort"] = effort
+    report_channel = record.get("slack_report_channel")
+    if isinstance(report_channel, str) and report_channel.strip():
+        configurable["slack_thread"] = {"channel_id": report_channel.strip()}
     return {"configurable": configurable, "metadata": _agent_version_metadata()}
 
 

agent/dispatch.py

@@ -62,12 +62,15 @@ async def dispatch_agent_run(
     assistant_id: str = "agent",
     metadata: dict[str, Any] | None = None,
     client: LangGraphClient | None = None,
+    multitask_strategy: str = "interrupt",
 ) -> dict[str, Any]:
     """Create (or interrupt-and-resume) a run for ``thread_id``.
 
     Routes every Slack / Linear / GitHub / dashboard trigger through one
     contract. ``source`` is for logging/metadata only; ``assistant_id`` selects
-    the graph (``"agent"`` or ``"reviewer"``).
+    the graph (``"agent"`` or ``"reviewer"``). ``multitask_strategy`` defaults to
+    ``"interrupt"`` (human follow-ups halt + resume); autofix passes ``"reject"``
+    so a burst of concurrent CI events for one head SHA can't interrupt each other.
     """
     client = client or dispatch_client()
     run = await client.runs.create(
@@ -75,7 +78,7 @@ async def dispatch_agent_run(
         assistant_id,
         input={"messages": [{"role": "user", "content": content}]},
         config={"configurable": configurable, "metadata": metadata or {}},
-        multitask_strategy="interrupt",
+        multitask_strategy=multitask_strategy,
         durability="sync",
         webhook=COMPLETION_WEBHOOK_URL,
         if_not_exists="create",

agent/prompt.py

@@ -368,6 +368,23 @@ ALWAYS_CREATE_PR_SECTION = """---
 The user's dashboard setting **Always Create PRs** is enabled. For code-change tasks, always open or update a draft pull request after committing and pushing the branch. This does not apply to questions, explanations, status checks, or other information-only requests where no files are changed."""
 
 
+def _render_scheduled_report_section(channel_id: str | None) -> str:
+    if not channel_id or not channel_id.strip():
+        return ""
+    return (
+        "---\n\n"
+        "### Scheduled Run Report\n\n"
+        "This is a scheduled (automated) run with a configured Slack report channel. "
+        "When you finish, post your final summary to that channel by calling "
+        "`slack_thread_reply` with your report — it posts a top-level message to the "
+        f"configured channel (`{channel_id.strip()}`) as the bot. Post exactly one "
+        "final report. If the run produced a pull request, include its link. If "
+        "`slack_thread_reply` reports a failure (for example `not_in_channel`, meaning "
+        "the bot is not a member of the channel), do not retry repeatedly — surface the "
+        "error in your final output instead."
+    )
+
+
 def _render_repo_instructions_section(instructions: str | None) -> str:
     if not instructions or not instructions.strip():
         return ""
@@ -398,6 +415,7 @@ SYSTEM_PROMPT_TEMPLATE = (
     + EXTERNAL_UNTRUSTED_COMMENTS_SECTION
     + COMMIT_PR_SECTION
     + "{pr_policy_override_section}"
+    + "{scheduled_report_section}"
     + "{collaboration_section}"
     + "{repo_instructions_section}"
 )
@@ -415,6 +433,7 @@ def construct_system_prompt(
     repo_custom_instructions: str | None = None,
     thread_url: str | None = None,
     corridor_enabled: bool = False,
+    slack_report_channel: str | None = None,
 ) -> str:
     default_prompt_section = _load_default_prompt()
     if default_repo and default_repo.get("owner") and default_repo.get("name"):
@@ -444,6 +463,7 @@ def construct_system_prompt(
         default_prompt_section=default_prompt_section,
         corridor_prompt_section=CORRIDOR_PROMPT if corridor_enabled else "",
         pr_policy_override_section=ALWAYS_CREATE_PR_SECTION if create_prs else "",
+        scheduled_report_section=_render_scheduled_report_section(slack_report_channel),
         collaboration_section=_render_collaboration_section(triggering_user_identity, thread_url),
         repo_instructions_section=_render_repo_instructions_section(repo_custom_instructions),
         commit_identity_name=commit_identity_name,

agent/server.py

@@ -647,6 +647,17 @@ def _get_cached_sandbox_backend(thread_id: str) -> SandboxBackendProtocol:
     return sandbox_backend
 
 
+def _scheduled_report_channel(configurable: dict[str, Any]) -> str | None:
+    """Slack channel a scheduled run should post its final report to, if configured."""
+    if (configurable or {}).get("source") != "schedule":
+        return None
+    slack_thread = (configurable or {}).get("slack_thread") or {}
+    channel_id = slack_thread.get("channel_id")
+    if isinstance(channel_id, str) and channel_id.strip():
+        return channel_id.strip()
+    return None
+
+
 async def _observability_authorized(config: RunnableConfig, profile_login: str | None) -> bool:
     """Whether the triggering user may use the team observability tools.
 
@@ -928,6 +939,7 @@ async def get_agent(config: RunnableConfig) -> Pregel:
             repo_custom_instructions=repo_custom_instructions,
             thread_url=dashboard_thread_url(thread_id),
             corridor_enabled=bool(corridor_tools),
+            slack_report_channel=_scheduled_report_channel(configurable),
         ),
         tools=[
             http_request,

agent/tools/slack_thread_reply.py

@@ -1,5 +1,6 @@
 import json
 import os
+from collections import OrderedDict
 from typing import Any
 
 from langgraph.config import get_config
@@ -8,6 +9,7 @@ from langgraph_sdk import get_client
 from ..utils.slack import (
     convert_mentions_to_slack_format,
     post_slack_thread_reply_with_ts,
+    post_slack_top_level_message_with_ts,
     store_slack_message_run_mapping,
 )
 
@@ -15,6 +17,12 @@ LANGGRAPH_URL = os.environ.get("LANGGRAPH_URL") or os.environ.get(
     "LANGGRAPH_URL_PROD", "http://localhost:2024"
 )
 
+# Runs that have already posted their single top-level (channel) message. Scheduled
+# runs seed `slack_thread` with a channel but no `thread_ts`, so every reply would
+# otherwise spray a new top-level message into the report channel; cap it at one.
+_MAX_TRACKED_RUNS = 2048
+_top_level_posts: "OrderedDict[str, None]" = OrderedDict()
+
 
 async def slack_thread_reply(
     message: str,
@@ -50,17 +58,33 @@ async def slack_thread_reply(
 
     channel_id = slack_thread.get("channel_id")
     thread_ts = slack_thread.get("thread_ts")
-    if not channel_id or not thread_ts:
+    if not channel_id:
         return {
             "success": False,
-            "error": "Missing slack_thread.channel_id or slack_thread.thread_ts in config",
+            "error": "Missing slack_thread.channel_id in config",
         }
 
     if not message.strip():
         return {"success": False, "error": "Message cannot be empty"}
 
+    top_level = not thread_ts
+    run_key = _run_key(config) if top_level else None
+    if top_level and run_key is not None and run_key in _top_level_posts:
+        return {
+            "success": False,
+            "error": "A message was already posted to this channel for this run",
+            "hint": (
+                "Only one top-level message per run is allowed for the configured "
+                "report channel; post a single final report and do not call this again."
+            ),
+        }
+
     message = convert_mentions_to_slack_format(message)
-    if plan_approval:
+    if top_level:
+        # Interactive blocks (options / plan_approval) render dead buttons in a
+        # report channel where no run is driving the approval/option flow.
+        slack_blocks = blocks
+    elif plan_approval:
         slack_blocks = _build_plan_approval_blocks(message)
     else:
         slack_blocks = blocks or _build_option_blocks(message, options)
@@ -75,9 +99,25 @@ async def slack_thread_reply(
             "message_chars": len(message),
             "hint": _slack_reply_failure_hint(slack_error),
         }
+    if top_level and run_key is not None:
+        _top_level_posts[run_key] = None
+        if len(_top_level_posts) > _MAX_TRACKED_RUNS:
+            _top_level_posts.popitem(last=False)
     return {"success": True}
 
 
+def _run_key(config: dict[str, Any]) -> str | None:
+    candidates = [config.get("run_id")]
+    configurable = config.get("configurable")
+    if isinstance(configurable, dict):
+        candidates.append(configurable.get("run_id"))
+        candidates.append(configurable.get("thread_id"))
+    for candidate in candidates:
+        if isinstance(candidate, str) and candidate:
+            return candidate
+    return None
+
+
 def _build_option_blocks(message: str, options: list[str] | None) -> list[dict[str, Any]] | None:
     if not options:
         return None
@@ -188,11 +228,15 @@ def _slack_reply_failure_hint(slack_error: str | None) -> str:
 
 async def _post_and_store_mapping(
     channel_id: str,
-    thread_ts: str,
+    thread_ts: str | None,
     message: str,
     *,
     blocks: list[dict[str, Any]] | None = None,
 ) -> tuple[str | None, str | None]:
+    if not thread_ts:
+        # Top-level report posts are fire-and-forget: a scheduled run is one-shot, so
+        # there is no live run to route channel replies back to (no mapping stored).
+        return await post_slack_top_level_message_with_ts(channel_id, message, blocks=blocks)
     message_ts, slack_error = await post_slack_thread_reply_with_ts(
         channel_id, thread_ts, message, blocks=blocks
     )

tests/test_agent_schedules.py

@@ -136,6 +136,96 @@ def test_cron_validation_accepts_steps_ranges_and_lists() -> None:
     assert body.schedule == "*/15 9-17 * * 1,3,5"
 
 
+def test_slack_report_channel_normalizes_and_validates() -> None:
+    body = ScheduleCreateBody(
+        prompt="hello", schedule="0 9 * * 1", slack_report_channel=" #C0123ABCD "
+    )
+    assert body.slack_report_channel == "C0123ABCD"
+
+    blank = ScheduleCreateBody(prompt="hello", schedule="0 9 * * 1", slack_report_channel="   ")
+    assert blank.slack_report_channel is None
+
+    with pytest.raises(ValidationError):
+        ScheduleCreateBody(prompt="hello", schedule="0 9 * * 1", slack_report_channel="not a chan")
+
+    with pytest.raises(ValidationError):
+        ScheduleCreateBody(prompt="hello", schedule="0 9 * * 1", slack_report_channel="123456")
+
+
+async def test_create_agent_schedule_persists_slack_report_channel(fake_client, auth) -> None:  # noqa: ANN001, ARG001
+    body = ScheduleCreateBody(
+        name="Daily report",
+        prompt="Summarize merged PRs",
+        schedule="0 9 * * 1-5",
+        slack_report_channel="C0123ABCD",
+    )
+
+    result = await schedules.create_agent_schedule("alice", body, email="alice@example.com")
+
+    assert result["slackReportChannel"] == "C0123ABCD"
+    stored = fake_client.store.items[(tuple(schedules.SCHEDULES_NAMESPACE), result["id"])]
+    assert stored["slack_report_channel"] == "C0123ABCD"
+
+
+async def test_update_agent_schedule_clears_slack_report_channel(fake_client) -> None:  # noqa: ANN001
+    record = {
+        "id": "sched_1",
+        "name": "Daily",
+        "prompt": "Run daily",
+        "schedule": "0 9 * * *",
+        "repo": None,
+        "model": "Default",
+        "effort": None,
+        "slack_report_channel": "C0123ABCD",
+        "enabled": True,
+        "cron_id": "cron_old",
+        "created_by": "alice",
+        "user_email": "alice@example.com",
+        "created_at": "2026-01-01T00:00:00+00:00",
+        "updated_at": "2026-01-01T00:00:00+00:00",
+    }
+    await fake_client.store.put_item(schedules.SCHEDULES_NAMESPACE, "sched_1", record)
+
+    result = await schedules.update_agent_schedule(
+        "sched_1",
+        "alice",
+        ScheduleUpdateBody(slack_report_channel=""),
+        email="alice@example.com",
+    )
+
+    assert result["slackReportChannel"] is None
+
+
+def test_agent_run_config_seeds_slack_thread_channel() -> None:
+    record = {
+        "id": "sched_1",
+        "model": "Default",
+        "effort": None,
+        "created_by": "alice",
+        "user_email": "alice@example.com",
+        "slack_report_channel": "C0123ABCD",
+    }
+
+    config = schedules._agent_run_config(record, "thread_1")
+
+    assert config["configurable"]["slack_thread"] == {"channel_id": "C0123ABCD"}
+
+
+def test_agent_run_config_omits_slack_thread_without_channel() -> None:
+    record = {
+        "id": "sched_1",
+        "model": "Default",
+        "effort": None,
+        "created_by": "alice",
+        "user_email": "alice@example.com",
+        "slack_report_channel": None,
+    }
+
+    config = schedules._agent_run_config(record, "thread_1")
+
+    assert "slack_thread" not in config["configurable"]
+
+
 async def test_create_agent_schedule_registers_scheduler_cron(fake_client, auth) -> None:  # noqa: ANN001, ARG001
     body = ScheduleCreateBody(
         name="Daily report",

tests/test_ci_autofix.py

@@ -89,6 +89,15 @@ async def test_dispatch_happy_path(happy: dict[str, Any]) -> None:
     happy["status_check"].assert_awaited()
 
 
+@pytest.mark.asyncio
+async def test_autofix_dispatch_uses_reject_strategy(happy: dict[str, Any]) -> None:
+    # A burst of concurrent CI events for one head SHA can slip past the busy-check
+    # before the dedupe SHA is recorded; dispatching with "reject" lets the platform
+    # drop the duplicate concurrent creates instead of interrupting each other.
+    await _run()
+    assert happy["runs_create"].await_args.kwargs["multitask_strategy"] == "reject"
+
+
 @pytest.mark.asyncio
 async def test_batches_when_thread_busy(happy: dict[str, Any], monkeypatch) -> None:
     monkeypatch.setattr(ci_autofix, "get_thread_active_status", AsyncMock(return_value=True))

tests/test_completion_webhook.py

@@ -98,6 +98,36 @@ async def test_missing_thread_id_is_ignored() -> None:
     assert result["status"] == "ignored"
 
 
+@pytest.mark.asyncio
+async def test_claims_flag_before_posting(monkeypatch: pytest.MonkeyPatch) -> None:
+    # Claim-then-post: the dedup flag must be set before the reply is posted so a
+    # retried/concurrent webhook can't double-post the canned failure message.
+    client = _FakeClient(_slack_metadata())
+    monkeypatch.setattr(completion, "langgraph_client", lambda: client)
+
+    async def _reply(*_args: Any, **_kwargs: Any) -> bool:
+        assert client.threads.updates == [{"failure_reply_posted": True}]
+        return True
+
+    monkeypatch.setattr(completion, "post_slack_thread_reply", AsyncMock(side_effect=_reply))
+
+    result = await completion.handle_run_completion({"thread_id": "t1", "status": "error"})
+    assert result["status"] == "ok"
+
+
+@pytest.mark.asyncio
+async def test_does_not_post_when_claim_fails(monkeypatch: pytest.MonkeyPatch) -> None:
+    client = _FakeClient(_slack_metadata())
+    client.threads.update = AsyncMock(side_effect=RuntimeError("boom"))
+    monkeypatch.setattr(completion, "langgraph_client", lambda: client)
+    reply = AsyncMock(return_value=True)
+    monkeypatch.setattr(completion, "post_slack_thread_reply", reply)
+
+    result = await completion.handle_run_completion({"thread_id": "t1", "status": "error"})
+    assert result["status"] == "error"
+    reply.assert_not_called()
+
+
 @pytest.mark.asyncio
 async def test_no_reply_channel_does_not_flag(monkeypatch: pytest.MonkeyPatch) -> None:
     client = _FakeClient({"source": "schedule"})

tests/test_slack_thread_reply_tool.py

@@ -169,3 +169,157 @@ async def test_slack_thread_reply_builds_option_blocks(monkeypatch: pytest.Monke
     assert actions["type"] == "actions"
     assert [button["text"]["text"] for button in actions["elements"]] == ["A", "B"]
     assert actions["elements"][0]["action_id"] == "open_swe_option_select"
+
+
+def _channel_only_config() -> dict[str, Any]:
+    return {"configurable": {"slack_thread": {"channel_id": "C9"}}}
+
+
+def _channel_run_config() -> dict[str, Any]:
+    return {
+        "configurable": {
+            "slack_thread": {"channel_id": "C9"},
+            "thread_id": "run-1",
+        }
+    }
+
+
+@pytest.fixture(autouse=True)
+def _reset_top_level_posts() -> Any:
+    slack_reply_tool._top_level_posts.clear()
+    yield
+    slack_reply_tool._top_level_posts.clear()
+
+
+async def test_slack_thread_reply_requires_channel_id(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(slack_reply_tool, "get_config", lambda: {"configurable": {}})
+
+    result = await slack_reply_tool.slack_thread_reply("hello")
+
+    assert result["success"] is False
+    assert result["error"] == "Missing slack_thread.channel_id in config"
+
+
+async def test_slack_thread_reply_posts_top_level_when_no_thread_ts(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    captured: dict[str, Any] = {}
+
+    async def fake_top_level(
+        channel_id: str,
+        text: str,
+        *,
+        blocks: list[dict[str, Any]] | None = None,
+    ) -> tuple[str | None, str | None]:
+        captured.update({"channel_id": channel_id, "text": text, "blocks": blocks})
+        return "3.0", None
+
+    async def fail_thread_reply(*args: Any, **kwargs: Any) -> tuple[str | None, str | None]:
+        raise AssertionError("should not post a thread reply without thread_ts")
+
+    monkeypatch.setattr(slack_reply_tool, "get_config", _channel_only_config)
+    monkeypatch.setattr(slack_reply_tool, "post_slack_top_level_message_with_ts", fake_top_level)
+    monkeypatch.setattr(slack_reply_tool, "post_slack_thread_reply_with_ts", fail_thread_reply)
+
+    result = await slack_reply_tool.slack_thread_reply("Scheduled report")
+
+    assert result == {"success": True}
+    assert captured["channel_id"] == "C9"
+    assert captured["text"] == "Scheduled report"
+
+
+async def test_slack_thread_reply_top_level_surfaces_not_in_channel(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    async def fake_top_level(
+        channel_id: str,
+        text: str,
+        *,
+        blocks: list[dict[str, Any]] | None = None,
+    ) -> tuple[str | None, str | None]:
+        return None, "not_in_channel"
+
+    monkeypatch.setattr(slack_reply_tool, "get_config", _channel_only_config)
+    monkeypatch.setattr(slack_reply_tool, "post_slack_top_level_message_with_ts", fake_top_level)
+
+    result = await slack_reply_tool.slack_thread_reply("Scheduled report")
+
+    assert result["success"] is False
+    assert result["slack_error"] == "not_in_channel"
+    assert "do not retry" in result["hint"]
+
+
+async def test_slack_thread_reply_top_level_drops_interactive_blocks(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    captured: dict[str, Any] = {}
+
+    async def fake_top_level(
+        channel_id: str,
+        text: str,
+        *,
+        blocks: list[dict[str, Any]] | None = None,
+    ) -> tuple[str | None, str | None]:
+        captured["blocks"] = blocks
+        return "3.0", None
+
+    monkeypatch.setattr(slack_reply_tool, "get_config", _channel_only_config)
+    monkeypatch.setattr(slack_reply_tool, "post_slack_top_level_message_with_ts", fake_top_level)
+
+    options_result = await slack_reply_tool.slack_thread_reply("Pick", options=["A", "B"])
+    assert options_result == {"success": True}
+    assert captured["blocks"] is None
+
+    approval_result = await slack_reply_tool.slack_thread_reply("Plan?", plan_approval=True)
+    assert approval_result == {"success": True}
+    assert captured["blocks"] is None
+
+
+async def test_slack_thread_reply_allows_only_one_top_level_post_per_run(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    calls = 0
+
+    async def fake_top_level(
+        channel_id: str,
+        text: str,
+        *,
+        blocks: list[dict[str, Any]] | None = None,
+    ) -> tuple[str | None, str | None]:
+        nonlocal calls
+        calls += 1
+        return "3.0", None
+
+    monkeypatch.setattr(slack_reply_tool, "get_config", _channel_run_config)
+    monkeypatch.setattr(slack_reply_tool, "post_slack_top_level_message_with_ts", fake_top_level)
+
+    first = await slack_reply_tool.slack_thread_reply("First")
+    second = await slack_reply_tool.slack_thread_reply("Second")
+
+    assert first == {"success": True}
+    assert second["success"] is False
+    assert "Only one top-level message per run" in second["hint"]
+    assert calls == 1
+
+
+async def test_slack_thread_reply_failed_top_level_post_does_not_consume_slot(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    results = iter([(None, "rate_limited"), ("3.0", None)])
+
+    async def fake_top_level(
+        channel_id: str,
+        text: str,
+        *,
+        blocks: list[dict[str, Any]] | None = None,
+    ) -> tuple[str | None, str | None]:
+        return next(results)
+
+    monkeypatch.setattr(slack_reply_tool, "get_config", _channel_run_config)
+    monkeypatch.setattr(slack_reply_tool, "post_slack_top_level_message_with_ts", fake_top_level)
+
+    first = await slack_reply_tool.slack_thread_reply("First")
+    second = await slack_reply_tool.slack_thread_reply("Retry")
+
+    assert first["success"] is False
+    assert second == {"success": True}

ui/src/components/agents/AutomationEditor.tsx

@@ -44,9 +44,7 @@ function scheduleToSelection(
     (model) =>
       model.id === schedule.model && model.efforts.includes(schedule.effort!)
   )
-  return supported
-    ? { modelId: schedule.model, effort: schedule.effort }
-    : null
+  return supported ? { modelId: schedule.model, effort: schedule.effort } : null
 }
 
 export function AutomationEditor({
@@ -73,6 +71,9 @@ export function AutomationEditor({
   )
   const [repo, setRepo] = useState<string | null>(schedule?.repo ?? null)
   const [enabled, setEnabled] = useState(schedule?.enabled ?? true)
+  const [slackReportChannel, setSlackReportChannel] = useState(
+    schedule?.slackReportChannel ?? ""
+  )
   // undefined = untouched (derive from the schedule / default as models load).
   const [selectionOverride, setSelectionOverride] = useState<
     ModelSelection | null | undefined
@@ -115,6 +116,7 @@ export function AutomationEditor({
           repo,
           model_id: modelId,
           effort,
+          slack_report_channel: slackReportChannel.trim() || null,
         },
         { onSuccess: () => navigate({ to: "/agents/automations" }) }
       )
@@ -132,6 +134,7 @@ export function AutomationEditor({
           model_id: modelId,
           effort,
           enabled,
+          slack_report_channel: slackReportChannel.trim() || null,
         },
       },
       { onSuccess: () => navigate({ to: "/agents/automations" }) }
@@ -264,6 +267,21 @@ export function AutomationEditor({
           </div>
         </div>
 
+        <SectionLabel>Report to Slack channel</SectionLabel>
+        <div className="rounded-xl border border-[var(--ui-border)] bg-[var(--ui-surface)] p-3">
+          <input
+            value={slackReportChannel}
+            onChange={(e) => setSlackReportChannel(e.target.value)}
+            placeholder="Channel ID (e.g. C0123ABCD)"
+            className="w-full bg-transparent font-mono text-sm text-[var(--ui-text)] outline-none placeholder:text-[var(--ui-text-dim)]"
+          />
+          <p className="mt-2 text-xs text-[var(--ui-text-dim)]">
+            When set, each run posts its final summary to this channel as the
+            Open SWE bot. The bot must be a member of the channel. Leave blank
+            for no Slack report.
+          </p>
+        </div>
+
         {errorMessage && (
           <p className="mt-4 text-xs text-[var(--ui-danger)]">{errorMessage}</p>
         )}

ui/src/lib/agents/api.ts

@@ -27,6 +27,7 @@ export interface ScheduleCreateRequest {
   repo?: string | null
   model_id?: string | null
   effort?: string | null
+  slack_report_channel?: string | null
 }
 
 export interface ScheduleUpdateRequest {
@@ -37,6 +38,7 @@ export interface ScheduleUpdateRequest {
   model_id?: string | null
   effort?: string | null
   enabled?: boolean | null
+  slack_report_channel?: string | null
 }
 
 export interface ThreadPrDiffFile {

ui/src/lib/agents/types.ts

@@ -164,6 +164,7 @@ export interface AgentSchedule {
   repo: string | null
   model: string
   effort?: string | null
+  slackReportChannel?: string | null
   enabled: boolean
   cronId?: string | null
   lastThreadId?: string | null