From 549a90f1ac8ffbcc40e610e2ab303a12114bdc1e Mon Sep 17 00:00:00 2001 From: amoussa1229 <166072409+amoussa1229@users.noreply.github.com> Date: Tue, 30 Jun 2026 21:21:38 +0000 Subject: [PATCH] Harden scheduled Slack report channel posting The no-thread_ts top-level path fired for every slack_thread_reply call during a scheduled run, spraying disconnected messages and dead interactive buttons into the report channel. Cap top-level posts at one per run and drop options/plan_approval blocks in that mode, so the mechanism (not just the prompt) enforces a single clean report. Also tighten the channel-ID regex to require a leading letter and document why top-level posts store no run mapping. Refs: #82 --- agent/dashboard/schedules.py | 2 +- agent/tools/slack_thread_reply.py | 43 ++++++++++++- tests/test_agent_schedules.py | 3 + tests/test_slack_thread_reply_tool.py | 92 +++++++++++++++++++++++++++ 4 files changed, 138 insertions(+), 2 deletions(-) diff --git a/agent/dashboard/schedules.py b/agent/dashboard/schedules.py index 5a92044d..8356035b 100644 --- a/agent/dashboard/schedules.py +++ b/agent/dashboard/schedules.py @@ -23,7 +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-Z0-9]{6,}$") +_SLACK_CHANNEL_ID_RE = re.compile(r"^[A-Z][A-Z0-9]{5,}$") class ScheduleCreateBody(BaseModel): diff --git a/agent/tools/slack_thread_reply.py b/agent/tools/slack_thread_reply.py index b2bd0385..e9d64974 100644 --- a/agent/tools/slack_thread_reply.py +++ b/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 @@ -16,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, @@ -60,8 +67,24 @@ async def slack_thread_reply( 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) @@ -76,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 @@ -195,6 +234,8 @@ async def _post_and_store_mapping( 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 diff --git a/tests/test_agent_schedules.py b/tests/test_agent_schedules.py index c2908695..df2509b3 100644 --- a/tests/test_agent_schedules.py +++ b/tests/test_agent_schedules.py @@ -148,6 +148,9 @@ def test_slack_report_channel_normalizes_and_validates() -> 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( diff --git a/tests/test_slack_thread_reply_tool.py b/tests/test_slack_thread_reply_tool.py index 8fa1683e..cf76d5ff 100644 --- a/tests/test_slack_thread_reply_tool.py +++ b/tests/test_slack_thread_reply_tool.py @@ -175,6 +175,22 @@ 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": {}}) @@ -231,3 +247,79 @@ async def test_slack_thread_reply_top_level_surfaces_not_in_channel( 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}