Skip to content

Bind saved-history fixtures to the retained owner before loading #1175

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
trissim merged 1 commit into from
Oct 23, 2026
Merged

Bind saved-history fixtures to the retained owner before loading #1175

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
37 changes: 24 additions & 13 deletions tests/textual/test_saved_history_coverage.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,10 +4,13 @@
(agent text is line-drawn now, so a live message has no pager of its own).
"""

from functools import partial

from agent_comms.session_file import RegistrySessionLocation
from agent_comms.threads import Thread
from support import journals
from toad.screens.main import MainScreen
from toad.tab_identity.kinds import MailboxTab
from toad.widgets.agent_response import AgentResponse
from toad.widgets.session_details import SessionDetails
from toad.widgets.transcript_history import TranscriptHistory
Expand All @@ -19,32 +22,40 @@ async def saved_history(workspace, ui):
"Canonical retained source owns coverage independently of painted resources.",
id=f"a{number}", parent=f"a{number - 1}" if number else None)
for number in range(8)))
workspace.comms.registry.declare(Thread("saved", frozenset(), str(workspace.project), session_file=RegistrySessionLocation.from_path(journal, workspace.comms.root)))
page = workspace.comms.transcripts.thread_transcript_page("saved")
owner = workspace.comms.registry.declare(Thread(
"saved", frozenset(), str(workspace.project),
session_file=RegistrySessionLocation.from_path(journal, workspace.comms.root),
))
root = str(workspace.comms.root)
await ui.app.session_navigation.new(
MailboxTab(owner.incarnation, root),
partial(MainScreen.from_thread, workspace.project, root, owner.incarnation),
original=(root, owner.incarnation),
)
conversation = ui.conversation
ui.attach_agent(ui.detached_agent())
conversation.transcript.received(page)
await ui.until(lambda: conversation.contents.query_children(TranscriptHistory), why="the saved history")
await ui.until(lambda: conversation.transcript.reports_coverage,
why="the retained owner's saved history")
page = workspace.comms.transcripts.thread_transcript_page(owner.name)
return page, conversation.contents.query_children(TranscriptHistory).first(), conversation.query_one(SessionDetails)


async def test_saved_history_is_unavailable_while_its_source_is_parked_and_returns_when_revalidated(workspace):
async with workspace.app(size=(110, 35)) as ui:
page, history, details = await saved_history(workspace, ui)
conversation = ui.conversation
await ui.until(lambda: "Saved history available" in details.title, why="saved history reported available")
await ui.until(lambda: "Saved history available" in str(details.title), why="saved history reported available")
children, pages = tuple(history.walk_children()), tuple(history.pages)

await history.retire_source(parked=True)
await conversation.transcript.reveal_retained(history)
await ui.until(lambda: "Saved history not loaded" in details.title, why="saved history reported not loaded")
await conversation.transcript.binding.reveal_retained(conversation.transcript, history)
await ui.until(lambda: "Saved history not loaded" in str(details.title), why="saved history reported not loaded")
assert not history.state.reports_coverage

conversation.transcript.received(page)
await ui.until(lambda: history.state.reports_coverage, why="coverage reported again")
await ui.until(lambda: "Saved history available" in details.title, why="saved history reported available again")
await ui.until(lambda: "Saved history available" in str(details.title), why="saved history reported available again")
assert "Saved history available" in details.overview_text
assert history in conversation.window.histories
assert history is conversation.transcript.binding.history
assert tuple(history.pages) == pages
assert tuple(history.walk_children()) == children

Expand All @@ -58,9 +69,9 @@ async def test_a_live_message_does_not_count_as_saved_history_coverage(workspace
f"Live paragraph {i} is long enough to need more than one page." for i in range(32))))
await ui.step()
await history.retire_source(parked=True)
await conversation.transcript.reveal_retained(history)
await ui.until(lambda: "Saved history not loaded" in details.title, why="saved history reported not loaded")
await conversation.transcript.binding.reveal_retained(conversation.transcript, history)
await ui.until(lambda: "Saved history not loaded" in str(details.title), why="saved history reported not loaded")
assert not conversation.transcript.reports_coverage
conversation.transcript.received(page)
await ui.until(lambda: "Saved history available" in details.title, why="saved history reported available")
await ui.until(lambda: "Saved history available" in str(details.title), why="saved history reported available")
assert tuple(history.pages) == pages
Loading