diff --git a/.claude/skills/initial-setup/SKILL.md b/.claude/skills/initial-setup/SKILL.md index aca4e7b..9c053ce 100644 --- a/.claude/skills/initial-setup/SKILL.md +++ b/.claude/skills/initial-setup/SKILL.md @@ -206,7 +206,7 @@ Always use `databricks secrets put-secret` which handles input securely. ### 5a. Full deploy -Run the end-to-end deployment (install deps, validate bundle, deploy, run migrations, start app): +Run the end-to-end deployment, takes around 5-10 minutes to start when the ap compute is off (install deps, validate bundle, deploy, run migrations, start app): ``` make full-deploy diff --git a/claw/core/slack_client.py b/claw/core/slack_client.py index 9e08bf0..b5b1d2e 100644 --- a/claw/core/slack_client.py +++ b/claw/core/slack_client.py @@ -157,6 +157,19 @@ async def reply_to_thread( return await post_message(ws, config, channel, text, thread_ts=thread_ts) +async def list_user_reactions(ws: Any, config: Any, limit: int = 100) -> list[dict]: + """Return items the authenticated user has reacted to (most recent first).""" + data = _slack_request( + ws, config.SLACK_UC_CONNECTION, "GET", "/reactions.list", + params={"count": str(limit), "full": "true"}, + ) + + if not data.get("ok"): + raise SlackApiError(data.get("error", "unknown_error")) + + return data.get("items", []) + + async def remove_reaction( ws: Any, config: Any, diff --git a/claw/core/slack_poller.py b/claw/core/slack_poller.py index 1ecaab9..15c4723 100644 --- a/claw/core/slack_poller.py +++ b/claw/core/slack_poller.py @@ -13,7 +13,7 @@ from claw.models import Session as SessionModel from claw.core.mcp_mapper import build_mcp_servers from claw.core.sessions import close_session, get_active_sessions, get_recent_messages, get_session, get_sessions_to_evict, is_message_handled, save_message, update_last_poll_ts, upsert_session -from claw.core.slack_client import fetch_history, fetch_thread_replies, open_self_dm, remove_reaction, reply_to_thread, resolve_identity +from claw.core.slack_client import fetch_history, fetch_thread_replies, list_user_reactions, open_self_dm, post_message, remove_reaction, reply_to_thread, resolve_identity logger = logging.getLogger(__name__) @@ -196,12 +196,10 @@ async def _poll_once(self) -> None: # --- Phase 3: Reaction trigger detection --- if self._config.CLAW_TRIGGER_REACTION: - channels = self._config.reaction_channels or {self._channel_id} - for ch in channels: - try: - await self._check_reaction_triggers(ch) - except Exception: - logger.exception("Error checking reaction triggers for channel %s", ch) + try: + await self._check_reaction_triggers() + except Exception: + logger.exception("Error checking reaction triggers") # --- Phase 4: Evict stale sessions --- with DbSession(self._db_engine) as db: @@ -235,47 +233,82 @@ async def _poll_once(self) -> None: # Reaction trigger detection # ------------------------------------------------------------------ - async def _check_reaction_triggers(self, channel: str) -> None: - """Scan recent messages in *channel* for the trigger reaction.""" + async def _check_reaction_triggers(self) -> None: + """Scan the authenticated user's recent reactions for the trigger emoji.""" trigger = self._config.CLAW_TRIGGER_REACTION - messages = await fetch_history( - self._ws, self._config, channel, self._last_seen_ts, - ) + allowed_channels = self._config.reaction_channels - for msg in messages: + items = await list_user_reactions(self._ws, self._config) + logger.debug("Reaction check: %d items from reactions.list", len(items)) + + for item in items: + if item.get("type") != "message": + logger.debug("Skipping non-message item type=%s", item.get("type")) + continue + + channel = item.get("channel", "") + if allowed_channels and channel not in allowed_channels: + logger.debug("Skipping channel %s not in allowlist", channel) + continue + + msg = item.get("message", {}) reactions = msg.get("reactions", []) + logger.debug( + "Reaction item: channel=%s msg_ts=%s reactions=%s", + channel, msg.get("ts"), [r["name"] for r in reactions], + ) for reaction in reactions: if reaction["name"] != trigger: + logger.debug("Skipping reaction %s != trigger %s", reaction["name"], trigger) continue if self._user_id not in reaction.get("users", []): + logger.debug( + "Skipping reaction %s: user_id %s not in users %s", + reaction["name"], self._user_id, reaction.get("users", []), + ) continue - # Found our trigger reaction — check dedup + # Found our trigger reaction — check dedup using original msg ts msg_ts = msg["ts"] with DbSession(self._db_engine) as db: existing = get_session(db, msg_ts) if existing is not None: + logger.debug("Skipping reaction on msg %s — session already exists (dedup)", msg_ts) continue - # Create session + # Post a kickoff message in the self-DM to start the thread + msg_text = msg.get("text", "") + kickoff_text = ( + f"Thread started from reaction message:\n> {msg_text}" + ) + kickoff_result = await post_message( + self._ws, self._config, self._channel_id, kickoff_text, + ) + thread_ts = kickoff_result.get("ts", msg_ts) + self._posted_ts.add(thread_ts) + + # Create session keyed on the kickoff message ts (lives in self-DM) + with DbSession(self._db_engine) as db: + upsert_session(db, thread_ts, self._channel_id) + + # Also record original msg_ts so we don't re-trigger on re-polls with DbSession(self._db_engine) as db: upsert_session(db, msg_ts, channel) - # Build synthetic message with context prefix - msg_text = msg.get("text", "") + # Build synthetic message for agent processing context = ( f"User <@{self._user_id}> reacted to start this thread.\n\n" f"Original message: {msg_text}" ) synthetic = { "text": context, - "ts": msg_ts, - "thread_ts": msg_ts, + "ts": thread_ts, + "thread_ts": thread_ts, "user": self._user_id, } await self._handle_message(synthetic) - # Remove the trigger reaction + # Remove the trigger reaction from the original channel try: await remove_reaction( self._ws, self._config, channel, trigger, msg_ts, diff --git a/tests/test_slack_poller.py b/tests/test_slack_poller.py index 4f42abc..77e0d44 100644 --- a/tests/test_slack_poller.py +++ b/tests/test_slack_poller.py @@ -78,11 +78,17 @@ async def test_calls_conversations_history_with_oldest(self): patch("claw.core.slack_poller.DbSession"): await poller._poll_once() - body = ws.api_client.do.call_args.kwargs["body"] - assert body["connection_name"] == "test-slack-conn" - assert body["method"] == "GET" - assert body["path"] == "/conversations.history" - params = json.loads(body["params"]) + # Find the conversations.history call (reactions.list may follow it) + history_call = None + for call in ws.api_client.do.call_args_list: + b = call.kwargs["body"] + if b.get("path") == "/conversations.history": + history_call = b + break + assert history_call is not None, "conversations.history was not called" + assert history_call["connection_name"] == "test-slack-conn" + assert history_call["method"] == "GET" + params = json.loads(history_call["params"]) assert params["channel"] == "C123" assert params["oldest"] == "99.0" @@ -2441,46 +2447,59 @@ def _make_poller(self): poller._last_seen_ts = "99.0" return poller - def _reaction_msg(self, text="hello world", ts="300.001", reaction_name="onit", users=None): - """Build a message with a reactions array.""" + def _reaction_item(self, text="hello world", ts="300.001", channel="C123", reaction_name="onit", users=None): + """Build a reactions.list item with message and channel.""" if users is None: users = ["U_BOT"] return { - "text": text, - "user": "U_OTHER", - "ts": ts, - "reactions": [{"name": reaction_name, "users": users}], + "type": "message", + "channel": channel, + "message": { + "text": text, + "user": "U_OTHER", + "ts": ts, + "reactions": [{"name": reaction_name, "users": users}], + }, } @pytest.mark.asyncio async def test_reaction_trigger_creates_session(self): - """Authenticated user's reaction creates a session and calls _handle_message.""" + """Authenticated user's reaction posts kickoff in self-DM and calls _handle_message.""" poller = self._make_poller() - msg = self._reaction_msg() + item = self._reaction_item() + kickoff_response = {"ok": True, "ts": "500.001"} - with patch("claw.core.slack_poller.fetch_history", new_callable=AsyncMock, return_value=[msg]), \ + with patch("claw.core.slack_poller.fetch_history", new_callable=AsyncMock, return_value=[]), \ + patch("claw.core.slack_poller.list_user_reactions", new_callable=AsyncMock, return_value=[item]), \ + patch("claw.core.slack_poller.post_message", new_callable=AsyncMock, return_value=kickoff_response) as mock_post, \ patch("claw.core.slack_poller.get_active_sessions", return_value=[]), \ patch("claw.core.slack_poller.get_sessions_to_evict", return_value=[]), \ patch("claw.core.slack_poller.get_session", return_value=None), \ patch("claw.core.slack_poller.upsert_session") as mock_upsert, \ - patch("claw.core.slack_poller.remove_reaction", new_callable=AsyncMock) as mock_remove, \ + patch("claw.core.slack_poller.remove_reaction", new_callable=AsyncMock), \ patch.object(poller, "_handle_message", new_callable=AsyncMock) as mock_handle, \ patch("claw.core.slack_poller.DbSession"): await poller._poll_once() - # _handle_message should have been called with the context prefix + # Kickoff message posted to self-DM + mock_post.assert_called_once() + assert mock_post.call_args[0][2] == "C123" # self._channel_id + + # _handle_message called with kickoff ts, not original msg ts mock_handle.assert_called() call_msg = mock_handle.call_args[0][0] assert "reacted to start this thread" in call_msg["text"] assert "hello world" in call_msg["text"] + assert call_msg["thread_ts"] == "500.001" @pytest.mark.asyncio async def test_reaction_trigger_ignores_other_user(self): """Reaction by a different user (not the bot) is ignored.""" poller = self._make_poller() - msg = self._reaction_msg(users=["U_SOMEONE_ELSE"]) + item = self._reaction_item(users=["U_SOMEONE_ELSE"]) - with patch("claw.core.slack_poller.fetch_history", new_callable=AsyncMock, return_value=[msg]), \ + with patch("claw.core.slack_poller.fetch_history", new_callable=AsyncMock, return_value=[]), \ + patch("claw.core.slack_poller.list_user_reactions", new_callable=AsyncMock, return_value=[item]), \ patch("claw.core.slack_poller.get_active_sessions", return_value=[]), \ patch("claw.core.slack_poller.get_sessions_to_evict", return_value=[]), \ patch.object(poller, "_handle_message", new_callable=AsyncMock) as mock_handle, \ @@ -2493,11 +2512,12 @@ async def test_reaction_trigger_ignores_other_user(self): async def test_reaction_trigger_dedup_existing_session(self): """If a session already exists for that ts, skip (dedup).""" poller = self._make_poller() - msg = self._reaction_msg() + item = self._reaction_item() existing_session = MagicMock() existing_session.thread_ts = "300.001" - with patch("claw.core.slack_poller.fetch_history", new_callable=AsyncMock, return_value=[msg]), \ + with patch("claw.core.slack_poller.fetch_history", new_callable=AsyncMock, return_value=[]), \ + patch("claw.core.slack_poller.list_user_reactions", new_callable=AsyncMock, return_value=[item]), \ patch("claw.core.slack_poller.get_active_sessions", return_value=[]), \ patch("claw.core.slack_poller.get_sessions_to_evict", return_value=[]), \ patch("claw.core.slack_poller.get_session", return_value=existing_session), \ @@ -2512,24 +2532,29 @@ async def test_reaction_trigger_disabled_when_empty(self): """CLAW_TRIGGER_REACTION="" disables reaction detection entirely.""" poller = self._make_poller() poller._config.CLAW_TRIGGER_REACTION = "" - msg = self._reaction_msg() + item = self._reaction_item() - with patch("claw.core.slack_poller.fetch_history", new_callable=AsyncMock, return_value=[msg]), \ + with patch("claw.core.slack_poller.fetch_history", new_callable=AsyncMock, return_value=[]), \ + patch("claw.core.slack_poller.list_user_reactions", new_callable=AsyncMock) as mock_list, \ patch("claw.core.slack_poller.get_active_sessions", return_value=[]), \ patch("claw.core.slack_poller.get_sessions_to_evict", return_value=[]), \ patch.object(poller, "_handle_message", new_callable=AsyncMock) as mock_handle, \ patch("claw.core.slack_poller.DbSession"): await poller._poll_once() + mock_list.assert_not_called() mock_handle.assert_not_called() @pytest.mark.asyncio async def test_reaction_trigger_removes_reaction(self): - """remove_reaction is called after triggering to clean up the emoji.""" + """remove_reaction is called on the ORIGINAL channel/message after triggering.""" poller = self._make_poller() - msg = self._reaction_msg() + item = self._reaction_item(channel="C_OTHER") + kickoff_response = {"ok": True, "ts": "500.001"} - with patch("claw.core.slack_poller.fetch_history", new_callable=AsyncMock, return_value=[msg]), \ + with patch("claw.core.slack_poller.fetch_history", new_callable=AsyncMock, return_value=[]), \ + patch("claw.core.slack_poller.list_user_reactions", new_callable=AsyncMock, return_value=[item]), \ + patch("claw.core.slack_poller.post_message", new_callable=AsyncMock, return_value=kickoff_response), \ patch("claw.core.slack_poller.get_active_sessions", return_value=[]), \ patch("claw.core.slack_poller.get_sessions_to_evict", return_value=[]), \ patch("claw.core.slack_poller.get_session", return_value=None), \ @@ -2540,16 +2565,19 @@ async def test_reaction_trigger_removes_reaction(self): await poller._poll_once() mock_remove.assert_called_once_with( - poller._ws, poller._config, "C123", "onit", "300.001", + poller._ws, poller._config, "C_OTHER", "onit", "300.001", ) @pytest.mark.asyncio async def test_reaction_trigger_context_prefix(self): - """Opening message includes context prefix with original message text.""" + """Kickoff message and agent prompt include original message text.""" poller = self._make_poller() - msg = self._reaction_msg(text="Please review this PR") + item = self._reaction_item(text="Please review this PR") + kickoff_response = {"ok": True, "ts": "500.001"} - with patch("claw.core.slack_poller.fetch_history", new_callable=AsyncMock, return_value=[msg]), \ + with patch("claw.core.slack_poller.fetch_history", new_callable=AsyncMock, return_value=[]), \ + patch("claw.core.slack_poller.list_user_reactions", new_callable=AsyncMock, return_value=[item]), \ + patch("claw.core.slack_poller.post_message", new_callable=AsyncMock, return_value=kickoff_response) as mock_post, \ patch("claw.core.slack_poller.get_active_sessions", return_value=[]), \ patch("claw.core.slack_poller.get_sessions_to_evict", return_value=[]), \ patch("claw.core.slack_poller.get_session", return_value=None), \ @@ -2559,8 +2587,32 @@ async def test_reaction_trigger_context_prefix(self): patch("claw.core.slack_poller.DbSession"): await poller._poll_once() + # Kickoff message posted to self-DM with original text + kickoff_text = mock_post.call_args[0][3] + assert "Please review this PR" in kickoff_text + + # Agent prompt includes context prefix call_msg = mock_handle.call_args[0][0] - assert f"User <@U_BOT> reacted to start this thread" in call_msg["text"] + assert "User <@U_BOT> reacted to start this thread" in call_msg["text"] assert "Please review this PR" in call_msg["text"] - assert call_msg["ts"] == "300.001" - assert call_msg["thread_ts"] == "300.001" + # Uses kickoff ts, not original msg ts + assert call_msg["ts"] == "500.001" + assert call_msg["thread_ts"] == "500.001" + + @pytest.mark.asyncio + async def test_reaction_trigger_channel_allowlist(self): + """Reactions in channels not in CLAW_REACTION_CHANNELS are skipped.""" + poller = self._make_poller() + poller._config.CLAW_REACTION_CHANNELS = "C_ALLOWED" + poller._config.reaction_channels = {"C_ALLOWED"} + item = self._reaction_item(channel="C_NOT_ALLOWED") + + with patch("claw.core.slack_poller.fetch_history", new_callable=AsyncMock, return_value=[]), \ + patch("claw.core.slack_poller.list_user_reactions", new_callable=AsyncMock, return_value=[item]), \ + patch("claw.core.slack_poller.get_active_sessions", return_value=[]), \ + patch("claw.core.slack_poller.get_sessions_to_evict", return_value=[]), \ + patch.object(poller, "_handle_message", new_callable=AsyncMock) as mock_handle, \ + patch("claw.core.slack_poller.DbSession"): + await poller._poll_once() + + mock_handle.assert_not_called()