import asyncio import logging logging.basicConfig(level=logging.DEBUG) from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine from sqlalchemy import select from deerflow.persistence.base import Base from deerflow.persistence.maf_roundtable import make_maf_roundtable_store from deerflow.persistence.maf_roundtable.model import MafRoundtableConversationEventRow, MafRoundtableConversationRow from app.maf_roundtable.conversation_service import MafConversationService class FakeMafConfig: enabled = True default_mode = 'structured' default_model_name = 'test-model' default_reasoning_effort = None default_manager_agent_id = 'mgr-agent' default_max_rounds = 8 default_write_policy = 'confirm_high_risk' admin_diagnostic_access = True async def main(): engine = create_async_engine('sqlite+aiosqlite://') async with engine.begin() as conn: import deerflow.persistence.maf_roundtable.model await conn.run_sync(Base.metadata.create_all) sf = async_sessionmaker(engine, expire_on_commit=False) store = make_maf_roundtable_store(sf) svc = MafConversationService(store, maf_config=FakeMafConfig()) conv = await svc.create_conversation(user_id='u1', title='Test') # Monkey-patch _publish_conversation_event to trace calls orig = svc._publish_conversation_event async def traced_publish(*args, **kwargs): print(f' _publish_conversation_event called: event_type={kwargs.get("event_type")}') result = await orig(*args, **kwargs) print(f' _publish_conversation_event returned: {result}') return result svc._publish_conversation_event = traced_publish try: result = await svc.send_message(conversation_id=conv['id'], user_id='u1', content='Hello') print(f'send_message result: {result}') except Exception as e: import traceback print(f'send_message ERROR: {type(e).__name__}: {e}') traceback.print_exc() async with sf() as session: events = (await session.execute(select(MafRoundtableConversationEventRow))).scalars().all() for e in events: print(f' event: seq={e.seq} type={e.event_type}') conv_row = (await session.execute( select(MafRoundtableConversationRow).where(MafRoundtableConversationRow.id == conv['id']) )).scalars().first() print(f'last_stream_seq={conv_row.last_stream_seq}') await engine.dispose() asyncio.run(main())