diff --git a/claw/core/slack_poller.py b/claw/core/slack_poller.py index 68bb736..963f778 100644 --- a/claw/core/slack_poller.py +++ b/claw/core/slack_poller.py @@ -233,6 +233,11 @@ async def _handle_message(self, message: dict) -> None: channel = self._channel_id text = message.get("text", "") + # Guard: skip if a response task is already running for this thread + if thread_ts in self._response_tasks: + logger.debug("Skipping message ts=%s — response task already running for thread %s", ts, thread_ts) + return + # Check for close command if text.strip().lower() in self._config.close_commands: await self._close_thread(thread_ts, channel) @@ -282,9 +287,13 @@ async def _handle_message(self, message: dict) -> None: formatted = "\n".join(f"{m.role.capitalize()}: {m.content}" for m in history) prompt = f"Previous conversation context (for continuity after restart):\n{formatted}\n\nCurrent message:\n{prompt}" + # Persist user message BEFORE dispatching to agent (dedup window fix) + msg_ts = message.get("ts", "") + with DbSession(self._db_engine) as db: + save_message(db, thread_ts, "user", text, message_ts=msg_ts or None) + # Send prompt and spawn background task for response collection await send_message(client, prompt) - msg_ts = message.get("ts", "") self._response_tasks[thread_ts] = asyncio.create_task( self._run_response_loop(thread_ts, channel, text, slack_ts=msg_ts) ) @@ -450,9 +459,8 @@ async def _run_response_loop(self, thread_ts: str, channel: str, original_text: if reply_ts: self._posted_ts.add(reply_ts) - # Persist user + assistant messages + # Persist assistant message (user message already saved in _handle_message) with DbSession(self._db_engine) as db: - save_message(db, thread_ts, "user", original_text, message_ts=slack_ts or None) save_message(db, thread_ts, "assistant", full_response) # Update session with new sdk_session_id diff --git a/openspec/changes/pre-dispatch-save-and-task-guard/spec.md b/openspec/changes/pre-dispatch-save-and-task-guard/spec.md new file mode 100644 index 0000000..b441a16 --- /dev/null +++ b/openspec/changes/pre-dispatch-save-and-task-guard/spec.md @@ -0,0 +1,42 @@ +# Pre-Dispatch Save & Response Task Guard + +## Why + +Two related dedup bugs exist in `SlackPoller._handle_message`: + +1. **Late save_message** — The user message is persisted in `_run_response_loop` + (after the full agent response). During the response window, + `is_message_handled()` returns False and the next poll re-triggers the + same message. + +2. **Task overwrite** — If a second message arrives for the same thread while + `_run_response_loop` is still running, the old task reference in + `_response_tasks` is silently overwritten. Both tasks run concurrently, + producing conflicting replies and DB writes. + +## What Changes + +### A — Save user message before dispatch (closes #79) + +- `claw/core/slack_poller.py` — In `_handle_message`, call + `save_message(db, thread_ts, "user", text, message_ts=ts)` BEFORE + `send_message(client, prompt)`. +- `claw/core/slack_poller.py` — In `_run_response_loop`, remove the + `save_message(…, "user", …)` call. Keep the assistant message save. + +### B — Guard against concurrent response tasks (closes #80) + +- `claw/core/slack_poller.py` — At the top of `_handle_message`, after + extracting `thread_ts` and before the close-command check, add: + `if thread_ts in self._response_tasks: return`. + +## Files Changed + +- `claw/core/slack_poller.py` — both fixes +- `tests/test_slack_poller.py` — new tests for pre-dispatch save and task guard + +## Impact + +- Backend: poller message handling +- Tests: new unit tests added to existing test file +- DB: no schema changes diff --git a/tests/test_slack_poller.py b/tests/test_slack_poller.py index 54c7c8b..d7e46f9 100644 --- a/tests/test_slack_poller.py +++ b/tests/test_slack_poller.py @@ -2230,3 +2230,191 @@ async def test_non_close_message_runs_agent(self): mock_create.assert_called_once() mock_send.assert_called_once() + + +# --------------------------------------------------------------------------- +# Pre-dispatch save — issue #79 +# --------------------------------------------------------------------------- + +class TestPreDispatchSave: + """save_message for the user message must happen BEFORE send_message.""" + + def _make_poller(self): + ws = _make_ws({"ok": True, "messages": []}) + cfg = _make_config() + cfg.close_commands = {"/close"} + cfg.CLAW_NAME = "claw" + cfg.ENABLE_PROGRESS_UPDATES = False + cfg.CLAW_REPLY_HEADER = False + cfg.DATABRICKS_USER_PAT = "fake-pat" + poller = SlackPoller(ws, cfg, _make_db_engine()) + poller._channel_id = "C123" + poller._user_id = "U_BOT" + return poller + + @pytest.mark.asyncio + async def test_save_message_called_before_send_message(self): + """save_message for user msg is called BEFORE send_message (agent dispatch).""" + poller = self._make_poller() + call_order = [] + + async def track_send(*a, **kw): + call_order.append("send_message") + + def track_save(*a, **kw): + call_order.append("save_message") + return MagicMock() + + with patch("claw.core.slack_poller.DbSession"), \ + patch("claw.core.slack_poller.get_session", return_value=None), \ + patch("claw.core.slack_poller.upsert_session"), \ + patch("claw.core.slack_poller.build_mcp_servers", return_value=[]), \ + patch("claw.core.slack_poller.create_client", new_callable=AsyncMock, return_value=AsyncMock()), \ + patch("claw.core.slack_poller.send_message", new_callable=AsyncMock, side_effect=track_send), \ + patch("claw.core.slack_poller.save_message", side_effect=track_save), \ + patch("claw.core.slack_poller.get_recent_messages", return_value=[]): + await poller._handle_message(_msg("hello", ts="200.001", thread_ts="200.001")) + + assert "save_message" in call_order, "save_message was never called in _handle_message" + assert "send_message" in call_order, "send_message was never called" + assert call_order.index("save_message") < call_order.index("send_message"), \ + f"save_message must be called before send_message, got: {call_order}" + + @pytest.mark.asyncio + async def test_response_loop_does_not_save_user_message(self): + """_run_response_loop should only save the assistant message, NOT the user message.""" + poller = self._make_poller() + + async def _aiter(): + yield "response text" + mock_wrapper = MagicMock() + mock_wrapper.progress_event_received = False + mock_wrapper.__aiter__ = lambda self: _aiter() + mock_wrapper.session_id = "sdk-1" + mock_wrapper.is_error = False + mock_wrapper.stop_reason = None + mock_wrapper.num_turns = 1 + mock_wrapper.total_cost_usd = 0.01 + mock_wrapper.duration_ms = 100 + + mock_client = MagicMock() + poller._clients["200.001"] = mock_client + + with patch("claw.core.slack_poller.iter_response", return_value=mock_wrapper), \ + patch("claw.core.slack_poller.reply_to_thread", new_callable=AsyncMock, return_value={"ts": "200.999"}), \ + patch("claw.core.slack_poller.save_message") as mock_save, \ + patch("claw.core.slack_poller.upsert_session"), \ + patch("claw.core.slack_poller.DbSession"): + await poller._run_response_loop("200.001", "C123", "original text", slack_ts="200.001") + + # Should only save assistant, not user + save_roles = [call.args[2] if len(call.args) > 2 else call.kwargs.get("role") for call in mock_save.call_args_list] + assert "user" not in save_roles, f"_run_response_loop should not save user message, but saved roles: {save_roles}" + assert "assistant" in save_roles, f"_run_response_loop should save assistant message, but saved roles: {save_roles}" + + @pytest.mark.asyncio + async def test_save_message_receives_correct_message_ts(self): + """save_message in _handle_message passes the Slack ts as message_ts.""" + poller = self._make_poller() + saved_kwargs = {} + + def capture_save(*args, **kwargs): + saved_kwargs.update({"args": args, "kwargs": kwargs}) + return MagicMock() + + with patch("claw.core.slack_poller.DbSession"), \ + patch("claw.core.slack_poller.get_session", return_value=None), \ + patch("claw.core.slack_poller.upsert_session"), \ + patch("claw.core.slack_poller.build_mcp_servers", return_value=[]), \ + patch("claw.core.slack_poller.create_client", new_callable=AsyncMock, return_value=AsyncMock()), \ + patch("claw.core.slack_poller.send_message", new_callable=AsyncMock), \ + patch("claw.core.slack_poller.save_message", side_effect=capture_save) as mock_save, \ + patch("claw.core.slack_poller.get_recent_messages", return_value=[]): + await poller._handle_message(_msg("hello", ts="200.001", thread_ts="200.001")) + + # save_message should be called with message_ts="200.001" + assert mock_save.called, "save_message was not called in _handle_message" + call_args = mock_save.call_args + # Check positional or keyword for message_ts + if len(call_args.args) >= 5: + assert call_args.args[4] == "200.001" + else: + assert call_args.kwargs.get("message_ts") == "200.001", \ + f"Expected message_ts='200.001', got {call_args}" + + +# --------------------------------------------------------------------------- +# Response task guard — issue #80 +# --------------------------------------------------------------------------- + +class TestResponseTaskGuard: + """_handle_message skips messages when a response task is already running.""" + + def _make_poller(self): + ws = _make_ws({"ok": True, "messages": []}) + cfg = _make_config() + cfg.close_commands = {"/close"} + cfg.CLAW_NAME = "claw" + cfg.ENABLE_PROGRESS_UPDATES = False + cfg.CLAW_REPLY_HEADER = False + cfg.DATABRICKS_USER_PAT = "fake-pat" + poller = SlackPoller(ws, cfg, _make_db_engine()) + poller._channel_id = "C123" + poller._user_id = "U_BOT" + return poller + + @pytest.mark.asyncio + async def test_skips_message_when_response_task_running(self): + """If _response_tasks[thread_ts] exists, _handle_message returns early.""" + poller = self._make_poller() + + # Simulate a running task for thread 200.001 + fake_task = MagicMock(spec=asyncio.Task) + poller._response_tasks["200.001"] = fake_task + + with patch("claw.core.slack_poller.DbSession"), \ + patch("claw.core.slack_poller.get_session") as mock_get, \ + patch("claw.core.slack_poller.create_client", new_callable=AsyncMock) as mock_create, \ + patch("claw.core.slack_poller.send_message", new_callable=AsyncMock) as mock_send: + await poller._handle_message(_msg("new msg", ts="200.002", thread_ts="200.001")) + + # Should not create client or send message + mock_create.assert_not_called() + mock_send.assert_not_called() + # The existing task should NOT be overwritten + assert poller._response_tasks["200.001"] is fake_task + + @pytest.mark.asyncio + async def test_processes_message_when_no_running_task(self): + """When no task exists for the thread, message is processed normally.""" + poller = self._make_poller() + + with patch("claw.core.slack_poller.DbSession"), \ + patch("claw.core.slack_poller.get_session", return_value=None), \ + patch("claw.core.slack_poller.upsert_session"), \ + patch("claw.core.slack_poller.build_mcp_servers", return_value=[]), \ + patch("claw.core.slack_poller.create_client", new_callable=AsyncMock, return_value=AsyncMock()) as mock_create, \ + patch("claw.core.slack_poller.send_message", new_callable=AsyncMock) as mock_send, \ + patch("claw.core.slack_poller.save_message"), \ + patch("claw.core.slack_poller.get_recent_messages", return_value=[]): + await poller._handle_message(_msg("hello", ts="200.001", thread_ts="200.001")) + + mock_create.assert_called_once() + mock_send.assert_called_once() + + @pytest.mark.asyncio + async def test_close_command_also_blocked_by_running_task(self): + """Even /close is blocked when a response task is running (guard is absolute).""" + poller = self._make_poller() + + # Simulate a running task + fake_task = MagicMock(spec=asyncio.Task) + poller._response_tasks["200.001"] = fake_task + + with patch("claw.core.slack_poller.DbSession"), \ + patch("claw.core.slack_poller.close_session") as mock_close, \ + patch("claw.core.slack_poller.reply_to_thread", new_callable=AsyncMock, return_value={"ts": "999.001"}): + await poller._handle_message(_msg("/close", ts="200.002", thread_ts="200.001")) + + # Guard fires before close-command check, so close_session is NOT called + mock_close.assert_not_called()