From 96015598644568d31bb3f45dbd170ac4ac0572c5 Mon Sep 17 00:00:00 2001 From: sisyfyZhang Date: Sat, 10 Oct 2026 01:08:59 +1100 Subject: [PATCH] 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 --- backend/tests/test_run_journal.py | 33 +++++++++++++++++++++++++++++++ 1 file changed, 33 insertions(+) diff --git a/backend/tests/test_run_journal.py b/backend/tests/test_run_journal.py index c708bc0350..1a465f4c94 100644 --- a/backend/tests/test_run_journal.py +++ b/backend/tests/test_run_journal.py @@ -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)