fix: pre-dispatch save and response task guard

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

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

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()