test(runtime): pin the flush() yield with a version-independent regression test (#6569)

The existing cross-thread test only goes red on Python 3.13+, while every
backend CI workflow pins 3.12, so removing the yield added in #6565 would
stay green. Record a middleware event from a raw thread, join it with no
intervening await so the call_soon_threadsafe hop is provably pending, then
close(flush=True): without the yield the event is dropped on every version.

Follows up the review on #6565.

Co-authored-by: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
sisyfyZhang 2026-10-10 01:08:59 +11:00 committed by GitHub
parent 215758818e
commit 9601559864
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194

View File

@ -70,6 +70,39 @@ async def test_cross_thread_middleware_events_are_serialized_on_owner_loop():
assert events[0]["content"]["changes"]["to_phase"] == "warned"
@pytest.mark.anyio
async def test_close_commits_cross_thread_event_still_pending_when_flush_starts():
"""A hop queued by a worker that already returned must survive close(flush=True).
The worker runs on a raw thread and is joined with no await in between, so
its ``call_soon_threadsafe`` hop is provably still pending when ``close()``
starts. Without the yield at the top of ``flush()`` the flush executes zero
awaits, detach runs first and ``_put`` drops the event on ``_closed``; this
pins the behaviour on every Python version, not only on 3.13+ where
``asyncio.to_thread`` can resume before the hop runs.
"""
store = MemoryRunEventStore()
journal = RunJournal("r-pending", "t-pending", store, flush_threshold=100)
def record_from_tool_worker() -> None:
journal.record_middleware(
"tool_progress",
name="ToolProgressMiddleware",
hook="wrap_tool_call",
action="warn",
changes={"from_phase": "active", "to_phase": "warned"},
)
worker = threading.Thread(target=record_from_tool_worker)
worker.start()
worker.join()
await journal.close(flush=True)
events = await store.list_events("t-pending", "r-pending")
assert [event["event_type"] for event in events] == ["middleware:tool_progress"]
def test_middleware_event_without_owner_loop_keeps_cross_thread_append():
store = MemoryRunEventStore()
journal = RunJournal("r-sync", "t-sync", store, flush_threshold=100)