mirror of
https://github.com/Sea-Haven-Industries/open-swe.git
synced 2026-10-07 16:19:09 +00:00
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
This commit is contained in:
parent
4d407b98a3
commit
549a90f1ac
4 changed files with 138 additions and 2 deletions
|
|
@ -23,7 +23,7 @@ SCHEDULES_NAMESPACE: list[str] = ["agent_schedules"]
|
||||||
_AGENT_ASSISTANT_ID = "agent"
|
_AGENT_ASSISTANT_ID = "agent"
|
||||||
_SCHEDULER_ASSISTANT_ID = "scheduler"
|
_SCHEDULER_ASSISTANT_ID = "scheduler"
|
||||||
_CRON_FIELD_RANGES = ((0, 59), (0, 23), (1, 31), (1, 12), (0, 7))
|
_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):
|
class ScheduleCreateBody(BaseModel):
|
||||||
|
|
|
||||||
|
|
@ -1,5 +1,6 @@
|
||||||
import json
|
import json
|
||||||
import os
|
import os
|
||||||
|
from collections import OrderedDict
|
||||||
from typing import Any
|
from typing import Any
|
||||||
|
|
||||||
from langgraph.config import get_config
|
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"
|
"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(
|
async def slack_thread_reply(
|
||||||
message: str,
|
message: str,
|
||||||
|
|
@ -60,8 +67,24 @@ async def slack_thread_reply(
|
||||||
if not message.strip():
|
if not message.strip():
|
||||||
return {"success": False, "error": "Message cannot be empty"}
|
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)
|
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)
|
slack_blocks = _build_plan_approval_blocks(message)
|
||||||
else:
|
else:
|
||||||
slack_blocks = blocks or _build_option_blocks(message, options)
|
slack_blocks = blocks or _build_option_blocks(message, options)
|
||||||
|
|
@ -76,9 +99,25 @@ async def slack_thread_reply(
|
||||||
"message_chars": len(message),
|
"message_chars": len(message),
|
||||||
"hint": _slack_reply_failure_hint(slack_error),
|
"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}
|
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:
|
def _build_option_blocks(message: str, options: list[str] | None) -> list[dict[str, Any]] | None:
|
||||||
if not options:
|
if not options:
|
||||||
return None
|
return None
|
||||||
|
|
@ -195,6 +234,8 @@ async def _post_and_store_mapping(
|
||||||
blocks: list[dict[str, Any]] | None = None,
|
blocks: list[dict[str, Any]] | None = None,
|
||||||
) -> tuple[str | None, str | None]:
|
) -> tuple[str | None, str | None]:
|
||||||
if not thread_ts:
|
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)
|
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(
|
message_ts, slack_error = await post_slack_thread_reply_with_ts(
|
||||||
channel_id, thread_ts, message, blocks=blocks
|
channel_id, thread_ts, message, blocks=blocks
|
||||||
|
|
|
||||||
|
|
@ -148,6 +148,9 @@ def test_slack_report_channel_normalizes_and_validates() -> None:
|
||||||
with pytest.raises(ValidationError):
|
with pytest.raises(ValidationError):
|
||||||
ScheduleCreateBody(prompt="hello", schedule="0 9 * * 1", slack_report_channel="not a chan")
|
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
|
async def test_create_agent_schedule_persists_slack_report_channel(fake_client, auth) -> None: # noqa: ANN001, ARG001
|
||||||
body = ScheduleCreateBody(
|
body = ScheduleCreateBody(
|
||||||
|
|
|
||||||
|
|
@ -175,6 +175,22 @@ def _channel_only_config() -> dict[str, Any]:
|
||||||
return {"configurable": {"slack_thread": {"channel_id": "C9"}}}
|
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:
|
async def test_slack_thread_reply_requires_channel_id(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||||
monkeypatch.setattr(slack_reply_tool, "get_config", lambda: {"configurable": {}})
|
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["success"] is False
|
||||||
assert result["slack_error"] == "not_in_channel"
|
assert result["slack_error"] == "not_in_channel"
|
||||||
assert "do not retry" in result["hint"]
|
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}
|
||||||
|
|
|
||||||
Loading…
Add table
Reference in a new issue