test_chat_mailbox_lifecycle_isolated.py
Rama main · réplica de lectura del repositorio original.
"""Multi-step mailbox journeys using actual handlers, route guards and SQLite.
No provider or AgentsServer is imported/started. Every ledger and identity is
synthetic; lifecycle entry points come from the existing AST-only fixtures.
"""
from __future__ import annotations
import asyncio
from copy import deepcopy
from types import SimpleNamespace
import unittest
from unittest.mock import AsyncMock
import chat_mailbox
import test_chat_mailbox_archived_sender_isolated as archived
import tests.test_chat_mailbox_runtime_isolated as runtime
class MailboxLifecycleTests(unittest.IsolatedAsyncioTestCase):
async def asyncSetUp(self):
self.case = archived.ArchivedSenderMailboxTests()
await self.case.asyncSetUp()
self._cleanups.extend(self.case._cleanups)
self.case._cleanups.clear()
async def read_page(self, request_id, after_seq=0):
return await self.case.ns['read_provider_chat_mailbox'](SimpleNamespace(
source_session_id='sender', request_id=request_id, after_seq=after_seq, limit=25,
), SimpleNamespace(owner='recipient'))
async def reopen(self):
ledger = self.case.ns['Ledger'](self.case.ledger.path)
await ledger.initialize()
self.case.ledger = ledger
self.case.ns['CROSS_CHAT'] = ledger
return ledger
async def test_busy_goal_backlog_retry_reconnect_pages_and_late_arrivals(self):
case = self.case
before = case.work_snapshot()
# Concurrent sends and retry storms stay messages, not eighty queued
# execution turns. The durable idempotency key owns message identity.
receipts = await asyncio.gather(*[
case.send(f'backlog-{index % 60}') for index in range(80)
])
self.assertEqual(len({row['message_id'] for row in receipts}), 60)
self.assertEqual(sum(row['duplicate'] for row in receipts), 20)
self.assertEqual(case.work_snapshot(), before)
self.assertEqual((await case.inbox())['senders'][0]['unread_count'], 60)
case.ns['schedule_next_queued_turn'].assert_not_called()
first = await self.read_page('backlog-snapshot')
self.assertEqual(len(first['messages']), 25)
self.assertTrue(first['has_more'])
late = await asyncio.gather(*[case.send(f'late-{index}') for index in range(7)])
late_ids = {row['message_id'] for row in late}
pending = await case.ledger.mailbox_call(
'list_messages', 'recipient', 'sender', [runtime.PAIR],
after_seq=first['next_after_seq'], limit=25, unread_only=True,
)
cancelled_id, deleted_id = [row['message_id'] for row in pending['messages'][:2]]
await case.ledger.mailbox_call('cancel_message', cancelled_id, now=runtime.NOW)
await case.ns['delete_chat_mailbox_message']('recipient', deleted_id)
await self.reopen()
case.issue_normal_recipient_capability('recipient-after-disconnect')
new_owner = case.work_snapshot()
replay = await self.read_page('backlog-snapshot')
self.assertTrue(replay['replayed'])
self.assertEqual(replay['messages'], first['messages'])
self.assertEqual(replay['read_id'], first['read_id'])
consumed = [row['message_id'] for row in first['messages']]
cursor = first['next_after_seq']
while cursor is not None:
page = await self.read_page('backlog-snapshot', cursor)
self.assertEqual(page['read_id'], first['read_id'])
self.assertEqual(page['snapshot_seq'], first['snapshot_seq'])
consumed.extend(row['message_id'] for row in page['messages'])
cursor = page['next_after_seq']
self.assertEqual(len(consumed), 58)
self.assertEqual(len(set(consumed)), 58)
self.assertFalse(set(consumed) & (late_ids | {cancelled_id, deleted_id}))
self.assertEqual((await case.inbox())['senders'][0]['unread_count'], 7)
fresh = await self.read_page('fresh-after-backlog')
self.assertEqual({row['message_id'] for row in fresh['messages']}, late_ids)
self.assertFalse(fresh['mail_pending'])
self.assertEqual((await case.inbox())['senders'], [])
self.assertEqual(case.work_snapshot(), new_owner)
self.assertEqual(case.ns['STORE'].sessions['recipient']['codex_goal']['status'], 'active')
with case.ledger._transaction() as connection:
self.assertEqual(connection.execute('SELECT COUNT(*) FROM chat_mailbox_messages').fetchone()[0], 67)
self.assertEqual(connection.execute('SELECT COUNT(*) FROM cross_chat_exchanges').fetchone()[0], 0)
case.assert_no_execution()
async def test_restart_recovers_unadmitted_wake_but_never_replays_admitted_attempt(self):
case = self.case
case.set_recipient('idle')
goal = deepcopy(case.ns['STORE'].sessions['recipient']['codex_goal'])
for index in range(12):
await case.send(f'pre-restart-{index}')
claim = await case.ledger.mailbox_call('claim_wake', 'recipient', [runtime.PAIR], now=runtime.NOW)
self.assertEqual(claim['through_seq'], 12)
self.assertEqual(claim['state'], 'reserved')
await self.reopen()
self.assertEqual(await case.ledger.mailbox_call('recover_wakes'), 1)
self.assertEqual(await case.ledger.mailbox_call('unread_targets'), ['recipient'])
case.enable_wake_admission()
await asyncio.gather(*[
case.ns['_start_next_queued_turn_locked']('recipient', admission_backend='codex')
for _ in range(8)
])
self.assertEqual(len(case.launches), 1)
self.assertEqual(case.wake_state()['state'], 'admitted')
self.assertNotEqual(case.wake_state()['claim_id'], claim['claim_id'])
self.assertEqual(case.ns['STORE'].sessions['recipient']['codex_goal'], goal)
# After a crash the previous execution may already have done work.
# Recover pending storage without silently launching it twice.
await self.reopen()
case.set_recipient('idle')
self.assertEqual(await case.ledger.mailbox_call('recover_wakes'), 0)
await case.ns['_start_next_queued_turn_locked']('recipient', admission_backend='codex')
self.assertEqual(len(case.launches), 1)
for row in await case.ledger.mailbox_envelopes():
self.assertIsNone(row['read_at'])
# A genuinely newer arrival may wake the idle recipient again.
await case.send('post-restart-new-message')
await case.ns['_start_next_queued_turn_locked']('recipient', admission_backend='codex')
self.assertEqual([row['through_seq'] for row in case.launches], [12, 13])
case.assert_no_execution()
async def test_independent_sqlite_workers_cancel_and_read_have_one_atomic_winner(self):
case = self.case
other = case.ns['Ledger'](case.ledger.path)
await other.initialize()
for index in range(12):
receipt = await case.send(f'raced-message-{index}')
message_id = receipt['message_id']
cancelled, page = await asyncio.gather(
other.mailbox_call('cancel_message', message_id, now=runtime.NOW),
self.read_page(f'race-read-{index}'), return_exceptions=True,
)
self.assertIsInstance(page, dict, page)
row = (await case.ledger.mailbox_envelopes(message_id=message_id))[0]
if cancelled is True:
self.assertEqual(page['messages'], [])
self.assertEqual(row['excluded_reason'], 'cancelled')
self.assertIsNone(row['read_at'])
else:
self.assertIsInstance(cancelled, chat_mailbox.MailboxConflict)
self.assertEqual([item['message_id'] for item in page['messages']], [message_id])
self.assertIsNotNone(row['read_at'])
self.assertIsNone(row['excluded_reason'])
replay = await case.send(f'raced-message-{index}')
self.assertTrue(replay['duplicate'])
self.assertEqual(replay['message_id'], message_id)
self.assertEqual(replay['state'], 'cancelled' if cancelled is True else 'read')
with case.ledger._transaction() as connection:
self.assertEqual(connection.execute('SELECT COUNT(*) FROM chat_mailbox_messages').fetchone()[0], 12)
case.assert_no_execution()
async def test_cancelled_retry_retains_identity_and_obeys_live_route_and_body_fences(self):
case = self.case
first = await case.send('cancelled-stable-key')
await case.ledger.mailbox_call('cancel_message', first['message_id'], now=runtime.NOW)
case.set_recipient('idle')
case.ns['schedule_next_queued_turn'].reset_mock()
replay = await case.send('cancelled-stable-key')
self.assertEqual((replay['message_id'], replay['state'], replay['duplicate'], replay['execution_started']),
(first['message_id'], 'cancelled', True, False))
case.ns['schedule_next_queued_turn'].assert_not_called()
for event in case.ns['append_cross_chat_event_once'].await_args_list[-2:]:
self.assertEqual(event.args[2], 'chat_conversation_message_cancelled')
changed_body = SimpleNamespace(
mode='async_route_v1', action='instruction', artifact_grants=[],
body='Different synthetic body', idempotency_key='cancelled-stable-key',
wait_for_response=False, response_timeout_seconds=None, reply_to_message_id=None,
)
with self.assertRaises(runtime.HTTPException) as conflict:
await case.ns['submit_provider_route_handoff'](runtime.ROUTE, changed_body, SimpleNamespace(owner='sender'))
self.assertEqual(conflict.exception.status_code, 409)
# A new explicit send is distinct; cancellation is not route revocation.
newer = await case.send('distinct-explicit-message')
self.assertNotEqual(newer['message_id'], first['message_id'])
self.assertFalse(newer['duplicate'])
self.assertEqual(newer['state'], 'unread')
case.ns['STORE'].sessions['sender']['_revoked_provider_cross_chat_route_ids'] = [runtime.ROUTE]
with self.assertRaises(runtime.HTTPException) as revoked:
await case.send('cancelled-stable-key')
self.assertEqual(revoked.exception.status_code, 403)
with case.ledger._transaction() as connection:
self.assertEqual(connection.execute('SELECT COUNT(*) FROM chat_mailbox_messages').fetchone()[0], 2)
case.assert_no_execution()
async def test_legacy_cancelled_and_failed_mailbox_still_reject_without_execution(self):
case = self.case
envelope_id = 'handoff_' + 'd' * 32
body = 'Synthetic legacy message'
await case.ledger.create_instruction(
envelope_id=envelope_id, source_session_id='sender', source_run_id='sender-run',
target_session_id='recipient', body=body, idempotency_key='legacy-cancelled',
authorization_kind='configured_route', authorization_route_id=runtime.ROUTE,
)
await case.ledger.update(envelope_id, status='cancelled')
case.ns['reserve_provider_route_handoff'] = AsyncMock(return_value=({
'envelope_id': envelope_id, 'source_run_id': 'sender-run',
'target_session_id': 'recipient',
}, True))
request = SimpleNamespace(
mode=None, action='instruction', artifact_grants=[], body=body,
idempotency_key='legacy-cancelled', wait_for_response=False,
response_timeout_seconds=None, reply_to_message_id=None,
)
with self.assertRaises(runtime.HTTPException) as rejected:
await case.ns['submit_provider_route_handoff'](runtime.ROUTE, request, SimpleNamespace(owner='sender'))
self.assertEqual(rejected.exception.status_code, 409)
failed = await case.send('failed-mailbox')
await case.ledger.update(failed['message_id'], status='failed')
with self.assertRaises(runtime.HTTPException) as rejected:
await case.send('failed-mailbox')
self.assertEqual(rejected.exception.status_code, 409)
case.assert_no_execution()
if __name__ == '__main__':
unittest.main()