Fix failed pager cleanup inheriting its dead worker context

src/toad/transcript_source_preparation.py

@@ -3,6 +3,7 @@
 
 from collections import deque
 from contextlib import AsyncExitStack
+from contextvars import Context
 from dataclasses import dataclass, replace
 from functools import partial
 from typing import TYPE_CHECKING
@@ -409,7 +410,11 @@ async def join():
                 except WorkerCancelled:
                     pass
 
-        return AwaitComplete(join())
+        # The join owns only the cancelled Worker handles, not the caller's
+        # worker or message pump. AwaitComplete eagerly schedules its coroutine;
+        # inheriting a failed pager's active_worker makes Worker.wait reject this
+        # independent join as a self-wait.
+        return Context().run(AwaitComplete, join())
 
     def resume_source(self) -> None:
         self._source_state = self._source_state.resume()

tests/textual/test_transcript_lifecycle.py

@@ -3,11 +3,15 @@
 from agent_comms.session_file import RegistrySessionLocation
 import asyncio
 from dataclasses import replace
-from unittest.mock import patch
-
+import pytest
+from textual.await_complete import AwaitComplete
+from textual.worker import DeadlockError, get_current_worker
 
 from support import journals
+from support.agents import RecordedEvents
 from support.textual_steps import saved_history
+from toad.core.source_events import TranscriptCoverage
+from toad.paths import get_log
 from toad.transcript_state import DetachedTranscript, LiveTranscript, ProvisionalTranscript
 
 
@@ -28,8 +32,8 @@ async def loader(**bounds):
         await ui.until(lambda: history.is_mounted, why="the history mounted")
         assert isinstance(history.state, ProvisionalTranscript)
         assert not history.state.can_draw
-        coverage = []
-        history.publish_core = coverage.append
+        coverage = RecordedEvents()
+        subscription = history.core_publications.subscribe(TranscriptCoverage, coverage.receive)
         history.publish_committed()
         assert not coverage, "source coverage escaped before the first painted frame"
         await ui.step()
@@ -58,6 +62,25 @@ async def pending_source_read():
         assert not history.state.can_draw
         history.resume_source()
         await ui.until(lambda: history.state.accepts_source_work, why="source work accepted after resume")
+        # A pager can initiate retirement from its own operation. Its independent
+        # join must not inherit that worker's self-wait identity.
+        retired: asyncio.Future[AwaitComplete] = asyncio.get_running_loop().create_future()
+
+        async def retire_from_source():
+            retired.set_result(history.retire_source(parked=True))
+
+        history.reserve_source_work().schedule(history, retire_from_source)
+        await (await retired)
+        assert not history.state.can_draw
+        history.resume_source()
+        await ui.until(lambda: history.state.accepts_source_work, why="source accepts work after its own retirement")
+
+        # Context isolation must not weaken a genuine worker self-wait.
+        async def reject_self_wait():
+            with pytest.raises(DeadlockError):
+                await get_current_worker().wait()
+
+        await ui.app.run_worker(reject_self_wait).wait()
         # The framework worker is cancelled during retirement; its completion
         # cannot restore the earlier live source.
         await history.retire_source()
@@ -70,6 +93,7 @@ async def pending_source_read():
         await removal
         await ui.until(lambda: not history.is_attached, why="the history removed")
         assert not history.state.can_draw
+        subscription.close()
     assert "Retain the source" in source.read_text()
 
 
@@ -78,23 +102,38 @@ async def test_unexpected_source_failure_is_shown_once_and_retires_the_history(w
 
     workspace.thread("peer", "one")
     workspace.comms.messaging.send("peer", "#one", "Visible before the failed read")
-    async with workspace.app() as ui:
-        await ui.open_channel("#one", actor="peer")
-        chat = ui.app.screen.query_one(CommsChatView)
-        history = chat.message_history
-        await ui.until(lambda: not history.undrawn and history.state.accepts_source_work,
-                       why="the initial message body")
-        error = RuntimeError("source read failed loudly")
-
-        async def failed_read():
-            raise error
-
-        with patch.object(history, "source_failed", wraps=history.source_failed) as failed:
-            history.reserve_source_work().schedule(history, failed_read)
-            await ui.until(lambda: failed.call_count == 1, why="the source failure shown")
-            assert failed.call_args.args == (error,)
+    loop = asyncio.get_running_loop()
+    previous_handler = loop.get_exception_handler()
+    unhandled: list[dict[str, object]] = []
+
+    def record_unhandled(loop: asyncio.AbstractEventLoop, context: dict[str, object]) -> None:
+        unhandled.append(context)
+
+    loop.set_exception_handler(record_unhandled)
+    try:
+        async with workspace.app() as ui:
+            await ui.open_channel("#one", actor="peer")
+            chat = ui.app.screen.query_one(CommsChatView)
+            history = chat.message_history
+            await ui.until(lambda: not history.undrawn and history.state.accepts_source_work,
+                           why="the initial message body")
+            error = RuntimeError("source read failed loudly")
+            before = set(get_log().glob("*.txt"))
+
+            async def failed_read():
+                raise error
+
+            worker = history.reserve_source_work().schedule(history, failed_read)
+            await ui.until(lambda: worker.is_finished, why="the failed pager finishes")
+            logs = set(get_log().glob("*.txt")) - before
+            assert len(logs) == 1
+            assert str(error) in next(iter(logs)).read_text()
             assert not history.state.accepts_source_work
             workspace.comms.messaging.send("peer", "#one", "A later observation must not retry")
             await ui.step(.2)
-            assert failed.call_count == 1
+            assert set(get_log().glob("*.txt")) - before == logs
+            assert worker not in ui.app.workers
             assert not history.state.accepts_source_work
+        assert not unhandled
+    finally:
+        loop.set_exception_handler(previous_handler)