From b5c54ba4e3900ac70c2189ddd874d9d0104f5178 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Fri, 23 Oct 2026 15:06:02 -0400 Subject: [PATCH 1/6] Build incoming fragments through their envelope owner and retire failed reads --- src/toad/transcript_source_preparation.py | 1 + src/toad/transcript_state.py | 2 +- src/toad/wire_history.py | 4 ++-- 3 files changed, 4 insertions(+), 3 deletions(-) diff --git a/src/toad/transcript_source_preparation.py b/src/toad/transcript_source_preparation.py index 38bf6bc61..e63f05381 100644 --- a/src/toad/transcript_source_preparation.py +++ b/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") diff --git a/src/toad/transcript_state.py b/src/toad/transcript_state.py index 4ec3706dd..ec9cac2b5 100644 --- a/src/toad/transcript_state.py +++ b/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 diff --git a/src/toad/wire_history.py b/src/toad/wire_history.py index 0ba1ca38b..148229efc 100644 --- a/src/toad/wire_history.py +++ b/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, From 21e0bb55c0120b8dc882b60d1c4d85de53f8e725 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Fri, 23 Oct 2026 15:10:13 -0400 Subject: [PATCH 2/6] Check fatal source errors are shown once and never rearmed --- tests/textual/test_transcript_lifecycle.py | 28 ++++++++++++++++++++++ 1 file changed, 28 insertions(+) diff --git a/tests/textual/test_transcript_lifecycle.py b/tests/textual/test_transcript_lifecycle.py index 73b13d675..61478208c 100644 --- a/tests/textual/test_transcript_lifecycle.py +++ b/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 @@ -72,3 +73,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 From 80c979117e5aa9f0e8700b9dbb68c2beffac869c Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Fri, 23 Oct 2026 15:13:58 -0400 Subject: [PATCH 3/6] Declare fixture threads through Thread after removal of Registration.new_thread --- tests/support/workspace.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/tests/support/workspace.py b/tests/support/workspace.py index 5038fd4e1..6e82e681b 100644 --- a/tests/support/workspace.py +++ b/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) From dc7232d3cde3d01382710edd5ac793a220f08283 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Fri, 23 Oct 2026 15:15:21 -0400 Subject: [PATCH 4/6] Delete assertions for the removed checkpoint_available mirror --- tests/textual/test_transcript_lifecycle.py | 3 --- 1 file changed, 3 deletions(-) diff --git a/tests/textual/test_transcript_lifecycle.py b/tests/textual/test_transcript_lifecycle.py index 61478208c..1ec9ad913 100644 --- a/tests/textual/test_transcript_lifecycle.py +++ b/tests/textual/test_transcript_lifecycle.py @@ -36,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() @@ -51,7 +50,6 @@ 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() # A parked tab cancels its running operation and the pending End. # Resuming accepts work on the same source, never a dead worker. @@ -59,7 +57,6 @@ 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") - assert history.checkpoint_available # The framework worker is cancelled during retirement; its completion # cannot restore the earlier live source. await history.retire_source() From 1965a4332fbd6a48cad5dd2da7626643ecc5f9dd Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Fri, 23 Oct 2026 15:16:51 -0400 Subject: [PATCH 5/6] Drive pending-tail lifecycle through the window End action --- tests/textual/test_transcript_lifecycle.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/textual/test_transcript_lifecycle.py b/tests/textual/test_transcript_lifecycle.py index 1ec9ad913..5913f8af1 100644 --- a/tests/textual/test_transcript_lifecycle.py +++ b/tests/textual/test_transcript_lifecycle.py @@ -50,7 +50,7 @@ async def pending_source_read(): assert history.state.can_draw assert history.state.reports_coverage assert not history.state.accepts_source_work - 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) From 378c30c75ed2fc04c880c5a736a042d11a76f0a3 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Fri, 23 Oct 2026 15:28:49 -0400 Subject: [PATCH 6/6] Lock agent-comms at the merge that adds IncomingTranscript.of wire_history.py calls IncomingTranscript.of, which first exists on agent-comms main at db3b35509 (#2879). The lock pinned 70174680, so an install from this head would raise AttributeError on every wire read. Co-Authored-By: Claude Opus 5.5 --- uv.lock | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/uv.lock b/uv.lock index 66d22583f..78b6e56be 100644 --- a/uv.lock +++ b/uv.lock @@ -21,7 +21,7 @@ wheels = [ [[package]] name = "agent-comms" version = "0.1.0" -source = { git = "https://github.com/OpenHCSDev/agent-comms.git?branch=main#7017468028ea2428ce71fe30261aa8d827aa222a" } +source = { git = "https://github.com/OpenHCSDev/agent-comms.git?branch=main#db3b35509029bff6c38218472972e864cf8f3a56" } dependencies = [ { name = "agent-client-protocol" }, { name = "aiohttp" },