diff --git a/openhands/automation/ingest.py b/openhands/automation/ingest.py index d4c10eb3..75f39749 100644 --- a/openhands/automation/ingest.py +++ b/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, diff --git a/openhands/automation/streams/slack.py b/openhands/automation/streams/slack.py index 046ef31f..67fe6e17 100644 --- a/openhands/automation/streams/slack.py +++ b/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 + ), ) diff --git a/tests/test_conversations.py b/tests/test_conversations.py index 56f12c56..6c37fc66 100644 --- a/tests/test_conversations.py +++ b/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 diff --git a/tests/test_ingest.py b/tests/test_ingest.py index 0c314b77..5e5c39a5 100644 --- a/tests/test_ingest.py +++ b/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(): diff --git a/tests/test_streams.py b/tests/test_streams.py index 0f5ba5da..560b9781 100644 --- a/tests/test_streams.py +++ b/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):