From 01e86ffdbd0f77c3dd6f4b03c91a596504620659 Mon Sep 17 00:00:00 2001 From: sharpnotebot Date: Thu, 15 Oct 2026 16:00:55 +0800 Subject: [PATCH] fix(gateway): admit completion wakes whose source differs only by message_id --- gateway/session_automation.py | 10 ++- .../test_automation_notification_category.py | 73 +++++++++++++++++++ 2 files changed, 82 insertions(+), 1 deletion(-) diff --git a/gateway/session_automation.py b/gateway/session_automation.py index 01ee23c7e7843..74ffd050c2dfa 100644 --- a/gateway/session_automation.py +++ b/gateway/session_automation.py @@ -106,6 +106,10 @@ def admission_display(payload): return card +def _route_identity(source): + return {key: value for key, value in source.to_dict().items() if key != 'message_id'} + + def snapshot_automation(authority, adapter, event, identity): runner = authority.runner if (not event.internal or event.message_type != MessageType.TEXT or event.is_command() @@ -131,11 +135,15 @@ def snapshot_automation(authority, adapter, event, identity): # Recheck only the source/provenance: a completed input's retained media is already released. restored = restore_native({'text': event.text, 'native_text_v1': { key: value for key, value in envelope.items() if key != 'media'}}, runner) - if (restored.source.to_dict() != event.source.to_dict() + # Synthetic wakes clear ``message_id`` (#52694, #123097) while the committed human source keeps + # it, so compare the routing/provenance identity, never the platform reply anchor. + if (_route_identity(restored.source) != _route_identity(event.source) or runner._adapter_for_source(restored.source) is not adapter): raise RuntimeStoreError('admission_conflict') provenance = deepcopy(envelope['provenance']) source = deepcopy(envelope['source']) + # Never anchor the wake to the stale human message; ``to_dict`` already omits a null anchor. + source.pop('message_id', None) # Producer timestamps and platform reply IDs change on retry. Neither belongs # in the identity/fingerprint of the same immutable completion. envelope = {'source': source, 'route': entry.session_key, diff --git a/tests/gateway/test_automation_notification_category.py b/tests/gateway/test_automation_notification_category.py index bff17f4debbd2..f28a634e3ced6 100644 --- a/tests/gateway/test_automation_notification_category.py +++ b/tests/gateway/test_automation_notification_category.py @@ -152,3 +152,76 @@ async def test_native_automation_category_roundtrip(tmp_path, monkeypatch, categ assert (await authority.admit_automation(adapter, event, 'notice')).admission_id == receipt.admission_id finally: authority.db.close() + + +_SOURCE = dict(chat_id='42', chat_type='dm', user_id='42', user_name='human') + + +async def _native_owner(tmp_path, monkeypatch, human_message_id): + from gateway.session_ingress_context import native_callback, register_transport_home + from plugins.platforms.discord.adapter import DiscordAdapter + runner, authority = await _authority(tmp_path, monkeypatch) + adapter = DiscordAdapter(PlatformConfig(enabled=True, token='***', typing_indicator=False)) + runner.adapters = {Platform.DISCORD: adapter} + runner._wire_adapter_handlers(adapter) + register_transport_home(runner, None, tmp_path) + monkeypatch.setenv('DISCORD_ALLOWED_USERS', '42') + source = adapter.build_source(**_SOURCE) + # The committed human turn keeps its platform reply anchor on the source. + human = MessageEvent(text='human', source=adapter.build_source(**_SOURCE, message_id=human_message_id), + message_id=human_message_id) + with native_callback(runner, human, tmp_path): + await authority.admit_native(human) + route = runner.session_store._generate_session_key(source) + metadata = {'gateway_session_key': route, 'gateway_session_id': runner.session_store.peek_session_id(route)} + return runner, authority, adapter, source, metadata + + +@pytest.mark.asyncio +async def test_native_completion_wake_admits_when_source_differs_only_by_message_id(tmp_path, monkeypatch): + """A synthetic wake clears ``message_id`` (#52694/#123097) while the committed human + envelope keeps it; the admission must still match the route and drop the stale anchor.""" + from gateway.session_envelope import restore_native + runner, authority, adapter, source, metadata = await _native_owner(tmp_path, monkeypatch, 'human-1') + try: + assert source.message_id is None + event = MessageEvent(text='notice', source=source, internal=True, metadata=dict(metadata)) + receipt = await authority.admit_automation(adapter, event, 'notice') + row = get_session_admission(authority.db, admission_id=receipt.admission_id) + assert row is not None + envelope = row['payload']['native_text_v1'] + assert 'message_id' not in envelope['source'] + assert restore_native(row['payload'], runner).source.message_id is None + assert (await authority.admit_automation(adapter, event, 'notice')).admission_id == receipt.admission_id + owner = authority.logical_owner(metadata['gateway_session_id']) + automation = [r for r in list_session_admissions(authority.db, session_id=owner, pending_only=False) + if (r['payload'].get('native_text_v1') or {}).get('automation')] + assert len(automation) == 1 + finally: + authority.db.close() + + +@pytest.mark.asyncio +@pytest.mark.parametrize('field', ['user_name', 'user_id', 'thread_id']) +async def test_native_completion_wake_refuses_other_source_drift(tmp_path, monkeypatch, field): + _runner, authority, adapter, _source, metadata = await _native_owner(tmp_path, monkeypatch, 'human-1') + try: + drifted = adapter.build_source(**{**_SOURCE, field: '7'}) + event = MessageEvent(text='notice', source=drifted, internal=True, metadata=dict(metadata)) + with pytest.raises(RuntimeStoreError, match='admission_conflict'): + await authority.admit_automation(adapter, event, 'notice') + finally: + authority.db.close() + + +@pytest.mark.asyncio +async def test_native_completion_wake_refuses_another_adapter(tmp_path, monkeypatch): + from plugins.platforms.discord.adapter import DiscordAdapter + _runner, authority, _adapter, source, metadata = await _native_owner(tmp_path, monkeypatch, 'human-1') + try: + stranger = DiscordAdapter(PlatformConfig(enabled=True, token='***', typing_indicator=False)) + event = MessageEvent(text='notice', source=source, internal=True, metadata=dict(metadata)) + with pytest.raises(RuntimeStoreError, match='admission_conflict'): + await authority.admit_automation(stranger, event, 'notice') + finally: + authority.db.close()