test: confirm agent gathers correct thread ts

Author: zimeg

tests/scenario_tests/test_events_agent.py

@@ -80,6 +80,54 @@ def handle_action(ack, agent: BoltAgent):
         assert response.status == 200
         assert_target_called()
 
+    def test_agent_thread_ts_from_event_in_thread(self):
+        """Agent gets thread_ts from event when in a thread."""
+        app = App(client=self.web_client)
+
+        state = {"thread_ts": None}
+
+        def assert_target_called():
+            count = 0
+            while state["thread_ts"] is None and count < 20:
+                sleep(0.1)
+                count += 1
+            assert state["thread_ts"] is not None
+
+        @app.event("app_mention")
+        def handle_mention(agent: BoltAgent):
+            state["thread_ts"] = agent._thread_ts
+
+        request = BoltRequest(body=app_mention_in_thread_body, mode="socket_mode")
+        response = app.dispatch(request)
+        assert response.status == 200
+        assert_target_called()
+        # Should use event.thread_ts (the thread root), not event.ts
+        assert state["thread_ts"] == "1111111111.111111"
+
+    def test_agent_thread_ts_falls_back_to_ts(self):
+        """Agent falls back to event.ts when not in a thread."""
+        app = App(client=self.web_client)
+
+        state = {"thread_ts": None}
+
+        def assert_target_called():
+            count = 0
+            while state["thread_ts"] is None and count < 20:
+                sleep(0.1)
+                count += 1
+            assert state["thread_ts"] is not None
+
+        @app.event("app_mention")
+        def handle_mention(agent: BoltAgent):
+            state["thread_ts"] = agent._thread_ts
+
+        request = BoltRequest(body=app_mention_event_body, mode="socket_mode")
+        response = app.dispatch(request)
+        assert response.status == 200
+        assert_target_called()
+        # Should fall back to event.ts since no thread_ts
+        assert state["thread_ts"] == "1234567890.123456"
+
     def test_agent_kwarg_emits_experimental_warning(self):
         app = App(client=self.web_client)
 
@@ -140,6 +188,18 @@ def build_payload(event: dict) -> dict:
     }
 )
 
+app_mention_in_thread_body = build_payload(
+    {
+        "type": "app_mention",
+        "user": "W222",
+        "text": "<@W111> hello in thread",
+        "ts": "2222222222.222222",
+        "thread_ts": "1111111111.111111",  # Thread root timestamp
+        "channel": "C111",
+        "event_ts": "2222222222.222222",
+    }
+)
+
 action_event_body = {
     "type": "block_actions",
     "user": {"id": "W222", "username": "test_user", "name": "test_user", "team_id": "T111"},

tests/scenario_tests_async/test_events_agent.py

@@ -86,6 +86,56 @@ async def handle_action(ack, agent: AsyncBoltAgent):
         assert response.status == 200
         await assert_target_called()
 
+    @pytest.mark.asyncio
+    async def test_agent_thread_ts_from_event_in_thread(self):
+        """Agent gets thread_ts from event when in a thread."""
+        app = AsyncApp(client=self.web_client)
+
+        state = {"thread_ts": None}
+
+        async def assert_target_called():
+            count = 0
+            while state["thread_ts"] is None and count < 20:
+                await asyncio.sleep(0.1)
+                count += 1
+            assert state["thread_ts"] is not None
+
+        @app.event("app_mention")
+        async def handle_mention(agent: AsyncBoltAgent):
+            state["thread_ts"] = agent._thread_ts
+
+        request = AsyncBoltRequest(body=app_mention_in_thread_body, mode="socket_mode")
+        response = await app.async_dispatch(request)
+        assert response.status == 200
+        await assert_target_called()
+        # Should use event.thread_ts (the thread root), not event.ts
+        assert state["thread_ts"] == "1111111111.111111"
+
+    @pytest.mark.asyncio
+    async def test_agent_thread_ts_falls_back_to_ts(self):
+        """Agent falls back to event.ts when not in a thread."""
+        app = AsyncApp(client=self.web_client)
+
+        state = {"thread_ts": None}
+
+        async def assert_target_called():
+            count = 0
+            while state["thread_ts"] is None and count < 20:
+                await asyncio.sleep(0.1)
+                count += 1
+            assert state["thread_ts"] is not None
+
+        @app.event("app_mention")
+        async def handle_mention(agent: AsyncBoltAgent):
+            state["thread_ts"] = agent._thread_ts
+
+        request = AsyncBoltRequest(body=app_mention_event_body, mode="socket_mode")
+        response = await app.async_dispatch(request)
+        assert response.status == 200
+        await assert_target_called()
+        # Should fall back to event.ts since no thread_ts
+        assert state["thread_ts"] == "1234567890.123456"
+
     @pytest.mark.asyncio
     async def test_agent_kwarg_emits_experimental_warning(self):
         app = AsyncApp(client=self.web_client)
@@ -147,6 +197,18 @@ def build_payload(event: dict) -> dict:
     }
 )
 
+app_mention_in_thread_body = build_payload(
+    {
+        "type": "app_mention",
+        "user": "W222",
+        "text": "<@W111> hello in thread",
+        "ts": "2222222222.222222",
+        "thread_ts": "1111111111.111111",  # Thread root timestamp
+        "channel": "C111",
+        "event_ts": "2222222222.222222",
+    }
+)
+
 action_event_body = {
     "type": "block_actions",
     "user": {"id": "W222", "username": "test_user", "name": "test_user", "team_id": "T111"},