Render stored wire messages through IncomingTranscript.of

src/toad/transcript_source_preparation.py

@@ -129,6 +129,7 @@ def schedule_source_work(self, work) -> Worker | None:
     def source_failed(self, error) -> None:
         from toad import paths
 
+        self.retire_source()
         paths.save_traceback(error, "History_failure")
         self.app.show_error(error, "History")
 

src/toad/transcript_state.py

@@ -402,7 +402,7 @@ async def execute[Result](self, owner, work: Callable[[], Awaitable[Result]]) ->
                 # This update could not use its original read. Keep the
                 # committed source; a new request owns any later advance.
                 return False
-            except (OSError, ValueError) as error:
+            except Exception as error:
                 owner.source_failed(error)
                 return False
 

src/toad/wire_history.py

@@ -201,7 +201,7 @@ def of(cls, message: Message, *, reader: str, **fields) -> WireFragment:
 
     @classmethod
     def from_message(cls, message, *, reader, **fields):
-        return cls((IncomingTranscript(message=message, reader=reader),), **fields)
+        return cls((IncomingTranscript.of(message, reader=reader),), **fields)
 
     @property
     def message(self) -> Message:
@@ -255,7 +255,7 @@ class MessageFragment(BusMessageFragment, WireFragment, roles=(ConversationRole,
     @classmethod
     def from_message(cls, message, *, reader, **fields):
         return cls(
-            (IncomingTranscript(message=message, reader=reader),),
+            (IncomingTranscript.of(message, reader=reader),),
             text_part=message.body,
             row=MessageRow(message, reader),
             **fields,

tests/support/workspace.py

@@ -10,6 +10,7 @@
 from support.agents import AppAgents
 from support.roots import Roots, installed_wire
 from support.ui import TextualUi
+from agent_comms.threads import Thread
 from toad.session_scope import AgentBinding, NoAgent
 
 if TYPE_CHECKING:
@@ -39,7 +40,7 @@ def thread(self, name: str, *tags: str, running: bool = False, title: str | None
         from agent_comms.thread_text import ThreadText
 
         identity = ProcessIdentity.capture(os.getpid()) if running else NoProcess()
-        self.comms.registry.declare(self.comms.registry.new_thread(
+        self.comms.registry.declare(Thread(
             name, frozenset(tags), str(self.project), title=ThreadText.external(title), process_identity=identity))
         return self.comms.registry.require(name)
 

tests/textual/test_transcript_lifecycle.py

@@ -3,6 +3,7 @@
 from agent_comms.session_file import RegistrySessionLocation
 import asyncio
 from dataclasses import replace
+from unittest.mock import patch
 
 
 from support import journals
@@ -35,7 +36,6 @@ async def loader(**bounds):
         assert isinstance(history.state, LiveTranscript)
         assert history.state.can_draw
         assert history.state.reports_coverage
-        assert history.checkpoint_available
 
         entered, release = asyncio.Event(), asyncio.Event()
 
@@ -50,15 +50,13 @@ async def pending_source_read():
         assert history.state.can_draw
         assert history.state.reports_coverage
         assert not history.state.accepts_source_work
-        assert not history.checkpoint_available
-        history.request_latest()
+        history.window.action_scroll_end()
         # A parked tab cancels its running operation and the pending End.
         # Resuming accepts work on the same source, never a dead worker.
         await history.retire_source(parked=True)
         assert not history.state.can_draw
         history.resume_source()
         await ui.until(lambda: history.state.accepts_source_work, why="source work accepted after resume")
-        assert history.checkpoint_available
         # The framework worker is cancelled during retirement; its completion
         # cannot restore the earlier live source.
         await history.retire_source()
@@ -72,3 +70,30 @@ async def pending_source_read():
         await ui.until(lambda: not history.is_attached, why="the history removed")
         assert not history.state.can_draw
     assert "Retain the source" in source.read_text()
+
+
+async def test_unexpected_source_failure_is_shown_once_and_retires_the_history(workspace):
+    from toad.widgets.comms_chat import CommsChatView
+
+    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,)
+            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 not history.state.accepts_source_work