feat(slack): continue replies in owned threads

openhands/automation/ingest.py

@@ -49,6 +49,10 @@ class AcceptedEvent:
     occurred_at: datetime | None = None
     # When set, persisted as the run's event_payload in place of `payload`.
     parsed_event: BaseModel | None = None
+    # Some transports can identify a possible follow-up but cannot prove from
+    # the payload alone that this service owns its subject. Such events may
+    # continue an existing subject, but must never open a new run.
+    existing_subject_only: bool = False
 
 
 @dataclass(frozen=True, slots=True)
@@ -82,7 +86,8 @@ async def accept_event(
     its subject's conversation, creating no run. A subject whose run is still
     queued has the turn folded into that run instead, so a burst cannot leave
     one subject with two runs. Anything else -- no subject, no run holding it,
-    one whose sandbox has gone -- falls back to creating a run.
+    one whose sandbox has gone -- falls back to creating a run unless the
+    transport marked the event as `existing_subject_only`.
 
     `request` and `session_factory` are both telemetry plumbing. Telemetry
     resolves its distinct id from the database, and HTTP callers supply that
@@ -199,6 +204,9 @@ async def accept_event(
                 )
                 continue
 
+        if event.existing_subject_only:
+            continue
+
         # How a later event on this subject finds this run's sandbox.
         run = await create_automation_run(
             automation,

openhands/automation/streams/slack.py

@@ -32,9 +32,9 @@
 
 logger = logging.getLogger("automation.streams.slack")
 
-# Widen once the transport is proven; every other event type is acked and
-# dropped.
-SUPPORTED_EVENT_TYPES = frozenset({"app_mention"})
+# `message` is narrowed below to human-authored thread replies; bare channel
+# traffic and every other event type are acked and dropped.
+SUPPORTED_EVENT_TYPES = frozenset({"app_mention", "message"})
 
 
 @dataclass
@@ -147,6 +147,10 @@ def accepted_event(self, envelope: dict[str, Any]) -> AcceptedEvent | None:
         if event.get("bot_id") or event.get("subtype") == "bot_message":
             return None
 
+        if event_key == "message":
+            if not event.get("thread_ts"):
+                return None
+
         return AcceptedEvent(
             source=self.source,
             event_key=event_key,
@@ -166,6 +170,14 @@ def accepted_event(self, envelope: dict[str, Any]) -> AcceptedEvent | None:
                 payload=envelope,
                 source_override=self.source,
             ),
+            # Slack reports the thread-root author in `parent_user_id`, not
+            # the author of the immediately preceding reply. In a human-rooted
+            # thread, durable subject ownership is the only safe proof that
+            # this is a follow-up to an automation conversation.
+            existing_subject_only=(
+                event_key == "message"
+                and event.get("parent_user_id") != self.bot_user_id
+            ),
         )
 
 

tests/test_conversations.py

@@ -547,6 +547,119 @@ async def test_two_mentions_in_one_thread_reach_the_same_conversation(
     assert len(await fetch_runs(async_session)) == 1
 
 
+@pytest.mark.asyncio
+async def test_human_rooted_slack_reply_reaches_the_owned_thread(
+    org_id, async_session, mock_authenticated_user, delivered_turns
+):
+    """Slack names the root author, so ownership comes from the prior run."""
+    automation = make_automation(
+        org_id,
+        mock_authenticated_user.user_id,
+        continuing_trigger(on=["app_mention", "message"]),
+    )
+    async_session.add(automation)
+    await async_session.commit()
+
+    first = await _mention(async_session, org_id, slack_envelope(), "Ev1")
+    await start_run(async_session, uuid.UUID(first.run_ids[0]))
+
+    reply = slack_envelope(
+        ts="1755000009.000900",
+        thread_ts="1755000000.000100",
+        text="Can you explain that?",
+    )
+    reply["event"]["type"] = "message"
+    reply["event"]["parent_user_id"] = "U456"
+    result = await accept_event(
+        org_id,
+        AcceptedEvent(
+            source="slack",
+            event_key="message",
+            payload=reply,
+            provider_event_id="Ev2",
+            existing_subject_only=True,
+        ),
+        async_session,
+    )
+
+    key = f"{TEAM}/C123/1755000000.000100"
+    assert result.run_ids == []
+    assert result.conversation_ids == [
+        expected_conversation(org_id, automation.id, key)
+    ]
+    assert len(await fetch_runs(async_session)) == 1
+    assert "Can you explain that?" in delivered_turns[0][1]
+
+
+@pytest.mark.asyncio
+async def test_human_rooted_slack_reply_cannot_claim_an_unowned_thread(
+    org_id, async_session, mock_authenticated_user, delivered_turns
+):
+    """A message in an unrelated Slack thread must not start an automation."""
+    automation = make_automation(
+        org_id,
+        mock_authenticated_user.user_id,
+        continuing_trigger(on=["app_mention", "message"]),
+    )
+    async_session.add(automation)
+    await async_session.commit()
+
+    reply = slack_envelope(
+        ts="1755000009.000900",
+        thread_ts="1755000000.000100",
+        text="Conversation between other people",
+    )
+    reply["event"]["type"] = "message"
+    reply["event"]["parent_user_id"] = "U456"
+    result = await accept_event(
+        org_id,
+        AcceptedEvent(
+            source="slack",
+            event_key="message",
+            payload=reply,
+            provider_event_id="Ev1",
+            existing_subject_only=True,
+        ),
+        async_session,
+    )
+
+    assert result.run_ids == []
+    assert result.conversation_ids == []
+    assert await fetch_runs(async_session) == []
+    assert delivered_turns == []
+
+
+@pytest.mark.asyncio
+async def test_follow_up_only_event_cannot_use_a_dispatch_run_trigger(
+    org_id, async_session, mock_authenticated_user, delivered_turns
+):
+    """The transport guard also applies when the matching trigger is unthreaded."""
+    async_session.add(
+        make_automation(
+            org_id,
+            mock_authenticated_user.user_id,
+            {"type": "event", "source": "slack", "on": "message"},
+        )
+    )
+    await async_session.commit()
+
+    result = await accept_event(
+        org_id,
+        AcceptedEvent(
+            source="slack",
+            event_key="message",
+            payload=slack_envelope(thread_ts="1755000000.000100"),
+            provider_event_id="Ev1",
+            existing_subject_only=True,
+        ),
+        async_session,
+    )
+
+    assert result.run_ids == []
+    assert result.conversation_ids == []
+    assert await fetch_runs(async_session) == []
+
+
 @pytest.mark.asyncio
 async def test_an_event_arriving_mid_run_continues_that_conversation(
     org_id, async_session, mock_authenticated_user, delivered_turns

tests/test_ingest.py

@@ -593,6 +593,7 @@ async def test_accepted_event_defaults():
     assert event.provider_event_id is None
     assert event.occurred_at is None
     assert event.parsed_event is None
+    assert event.existing_subject_only is False
 
 
 def test_dataclasses_are_frozen():

tests/test_streams.py

@@ -216,6 +216,36 @@ async def test_ignored_envelope_is_still_acked(provider):
     assert client.calls == ["ack:env-1"]
 
 
+@pytest.mark.asyncio
+async def test_reply_to_bot_message_is_acked_before_it_is_routed(provider):
+    client = FakeSocketClient()
+    payload = envelope(
+        type="message",
+        text="Can you explain that?",
+        thread_ts="1755000000.000100",
+        parent_user_id=BOT_USER_ID,
+    )
+
+    await provider.handle(client, request(payload), client.emit)
+
+    assert client.calls == ["ack:env-1", "emit:message"]
+
+
+@pytest.mark.asyncio
+async def test_human_rooted_thread_reply_is_acked_before_it_is_routed(provider):
+    client = FakeSocketClient()
+    payload = envelope(
+        type="message",
+        text="Can you explain that?",
+        thread_ts="1755000000.000100",
+        parent_user_id="U456",
+    )
+
+    await provider.handle(client, request(payload), client.emit)
+
+    assert client.calls == ["ack:env-1", "emit:message"]
+
+
 @pytest.mark.asyncio
 async def test_non_events_api_request_is_acked_only(provider):
     client = FakeSocketClient()
@@ -238,6 +268,55 @@ def test_app_mention_becomes_an_accepted_event(provider):
     assert event.provider_event_id == "Ev0001"
 
 
+def test_reply_to_bot_message_becomes_an_accepted_event(provider):
+    event = provider.accepted_event(
+        envelope(
+            type="message",
+            text="Can you explain that?",
+            thread_ts="1755000000.000100",
+            parent_user_id=BOT_USER_ID,
+        )
+    )
+
+    assert event is not None
+    assert event.source == "slack"
+    assert event.event_key == "message"
+    assert event.payload["event"]["parent_user_id"] == BOT_USER_ID
+    assert event.provider_event_id == "Ev0001"
+    assert event.existing_subject_only is False
+
+
+def test_human_rooted_thread_reply_requires_an_existing_subject(provider):
+    event = provider.accepted_event(
+        envelope(
+            type="message",
+            text="Can you explain that?",
+            thread_ts="1755000000.000100",
+            parent_user_id="U456",
+        )
+    )
+
+    assert event is not None
+    assert event.event_key == "message"
+    assert event.existing_subject_only is True
+
+
+@pytest.mark.parametrize(
+    "overrides",
+    [
+        {"type": "message"},
+        {
+            "type": "message",
+            "thread_ts": "1755000000.000100",
+            "parent_user_id": BOT_USER_ID,
+            "bot_id": "B123",
+        },
+    ],
+)
+def test_bare_or_bot_authored_messages_are_dropped(provider, overrides: dict[str, Any]):
+    assert provider.accepted_event(envelope(**overrides)) is None
+
+
 def test_the_run_payload_matches_the_webhook_path_exactly(provider):
     """Both transports must hand the run the identical structure.
 
@@ -284,7 +363,7 @@ def test_bot_messages_are_dropped(provider, overrides: dict):
 
 
 def test_other_event_types_are_dropped(provider):
-    assert provider.accepted_event(envelope(type="message")) is None
+    assert provider.accepted_event(envelope(type="reaction_added")) is None
 
 
 def test_other_teams_are_dropped(provider):