From fe7daf5330223069b0530bdda63bf05c1011ef90 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Sun, 1 Nov 2026 19:17:04 -0500 Subject: [PATCH 01/10] Keep captured thread identity through project changes and owner reloads --- src/agent_comms/thread_management.py | 12 +++--- src/agent_comms/thread_presentation.py | 16 +++++++- src/agent_comms/tools.py | 2 +- tests/test_bulk_lifetimes.py | 2 +- tests/test_checkout_submodules.py | 10 ++--- tests/test_project.py | 13 +++--- tests/test_project_resume.py | 8 ++-- tests/test_project_source_identity.py | 55 ++++++++++++++++++++++++++ tests/test_registration_inheritance.py | 8 ++-- tests/test_session_load_plan.py | 49 ++++++++++++++++++++++- 10 files changed, 146 insertions(+), 29 deletions(-) create mode 100644 tests/test_project_source_identity.py diff --git a/src/agent_comms/thread_management.py b/src/agent_comms/thread_management.py index a44a68fa8e..91449300f6 100644 --- a/src/agent_comms/thread_management.py +++ b/src/agent_comms/thread_management.py @@ -58,7 +58,7 @@ from .pi_vocabulary import NoThinkingOverride, ThinkingChoice, ThinkingLevelField from .registry_provenance import RegistryNames from .session_file import RegistrySessionLocation -from .thread_identity import AgentThreadRole, ThreadRole +from .thread_identity import AgentThreadRole, ThreadIncarnation, ThreadRole from .thread_model import DeclaredModel, ThreadModel, UndeclaredModel from .thread_owned_state import ThreadOwnedState from .thread_text import ForkedParent, GivenThreadText, NoThreadText, ThreadText @@ -354,11 +354,13 @@ def rename_self(self, new_name: str, *, caller: ThreadCaller) -> RenameThreadRes edit, caller.name, new_name, title=GivenThreadText(new_name.replace("-", " ").replace("_", " ")) ) - def set_project(self, name: str, path: str) -> ProjectChangeResult: - """Move one thread's project, preserving its owner and conversation.""" + def set_project(self, target_thread: ThreadIncarnation, path: str) -> ProjectChangeResult: + """Move the captured thread's project, never a later occupant of its name.""" if not path.strip(): raise ValueError("Project path cannot be empty.") - original = self.registry.require(name) + original = self.registry.require(target_thread.name) + if original.incarnation != target_thread: + raise RelationViolationError("Thread incarnation changed before project selection") target = Path(path).expanduser() if not target.is_absolute(): target = original.worktree.path / target @@ -374,7 +376,7 @@ def move(current: Thread) -> Thread: current.require_original_project(original) return current if current.project_path == target else current.moved_to_project(project) - updated = self.registry.update_thread(name, move) + updated = self.registry.update_thread(original.name, move) previous, current = str(original.project_path), str(updated.project_path) return ProjectChangeResult(updated.name, previous, current, current != previous) diff --git a/src/agent_comms/thread_presentation.py b/src/agent_comms/thread_presentation.py index b372018ab3..3ae429e7bf 100644 --- a/src/agent_comms/thread_presentation.py +++ b/src/agent_comms/thread_presentation.py @@ -2,7 +2,7 @@ from __future__ import annotations -from collections.abc import Callable +from collections.abc import Awaitable, Callable from dataclasses import dataclass, field, fields, replace from typing import TYPE_CHECKING, ClassVar, NoReturn, Self @@ -16,6 +16,7 @@ from .message_reference import MessageReference if TYPE_CHECKING: + from .session_load import SessionLoadPlan from .presentation import MessageNotification from .transcripts import AcquiredThreadPresentation, TranscriptReadIdentity from agent_comms.registry_document import RegistrySnapshot @@ -149,6 +150,11 @@ def replaces(self, original: LiveThreadOwnerBinding) -> bool: def superseded_by(self, replacement: ThreadOwnerBinding) -> bool: return False + async def reload_if_superseded( + self, previous: SessionLoadPlan, reload: Callable[[SessionLoadPlan], Awaitable[None]], + ) -> None: + """An unavailable owner cannot supply a new existing-session load.""" + def recipient_activity(self, activity, snapshot, thread): from .agent_activity import UnavailableRecipientActivity @@ -186,6 +192,14 @@ def recipient_activity(self, activity, snapshot, thread): def superseded_by(self, replacement: ThreadOwnerBinding) -> bool: return replacement.replaces(self) + async def reload_if_superseded( + self, previous: SessionLoadPlan, reload: Callable[[SessionLoadPlan], Awaitable[None]], + ) -> None: + from .session_load import ExistingSessionLoadPlan + + if previous.superseded_by(self): + await reload(ExistingSessionLoadPlan(self)) + def replaces(self, original: LiveThreadOwnerBinding) -> bool: # Both operands are the ORIGINAL registry owner/process witnesses. # Queued-input generations use a different counter. diff --git a/src/agent_comms/tools.py b/src/agent_comms/tools.py index f4693aa118..fcf00358ce 100644 --- a/src/agent_comms/tools.py +++ b/src/agent_comms/tools.py @@ -517,7 +517,7 @@ class CommsSetProjectTool(ThreadToolRequest): path: Annotated[str, *tool_field("New project directory")] def apply(self, comms: Comms, caller: ThreadCaller) -> ProjectSet: - result = comms.threads.set_project(caller.name, self.path) + result = comms.threads.set_project(caller.identity(comms), self.path) return ProjectSet( result.thread, result.previous, result.current, result.changed, ( diff --git a/tests/test_bulk_lifetimes.py b/tests/test_bulk_lifetimes.py index 6fc60d3726..f587392c4c 100644 --- a/tests/test_bulk_lifetimes.py +++ b/tests/test_bulk_lifetimes.py @@ -98,7 +98,7 @@ def test_rehome_refuses_a_changed_original_before_publishing_the_staged_copy( rehomed = workers.submit(comms.threads.rehome_thread, paused_import, name=original.name) try: assert paused_import.snapshot.entered.wait(5) - moved = workers.submit(comms.threads.set_project, original.name, str(changed_project)) + moved = workers.submit(comms.threads.set_project, original.incarnation, str(changed_project)) moved.result(timeout=2) finally: paused_import.snapshot.proceed.set() diff --git a/tests/test_checkout_submodules.py b/tests/test_checkout_submodules.py index fe32655668..3a93e75108 100644 --- a/tests/test_checkout_submodules.py +++ b/tests/test_checkout_submodules.py @@ -520,7 +520,7 @@ def test_transfer_refuses_current_reviewer_then_keeps_departed_reviewers_history assert (git(source.path, 'rev-parse', 'HEAD'), git(source.path, 'status', '--porcelain')) == before - assert comms.threads.set_project(reviewer.name, str(reviewer_path)).changed + assert comms.threads.set_project(reviewer.incarnation, str(reviewer_path)).changed departed = comms.registry.require(reviewer.name) assert departed.incarnation == reviewer.incarnation assert departed.worktree == Checkout(reviewer_path) @@ -548,8 +548,8 @@ def test_transfer_refuses_foreign_allocation_even_after_its_owner_changes_projec 'reviewer', frozenset(), source, execution=ExternalThreadExecution)) reviewer_path = rule.order_checkout(rule.roots_of(reviewer), 'P3') Checkout.add(repository, reviewer_path, NewBranchStart('reviewer-next', 'HEAD')) - assert comms.threads.set_project(worker.name, str(worker_path)).changed - assert comms.threads.set_project(reviewer.name, str(reviewer_path)).changed + assert comms.threads.set_project(worker.incarnation, str(worker_path)).changed + assert comms.threads.set_project(reviewer.incarnation, str(reviewer_path)).changed worker = comms.registry.require(worker.name) reviewer = comms.registry.require(reviewer.name) assert worker.worktree == Checkout(worker_path) @@ -581,8 +581,8 @@ def test_transfer_accepts_callers_historical_launch_without_current_allocation( rule.require_transfer_source( worker, source, comms.registry.snapshot().threads.values(), comms.goals.todos.get('P1').repository) - assert comms.threads.set_project(worker.name, str(source)).changed - assert comms.threads.set_project(worker.name, str(next_project)).changed + assert comms.threads.set_project(worker.incarnation, str(source)).changed + assert comms.threads.set_project(worker.incarnation, str(next_project)).changed worker = comms.registry.require(worker.name) assert checkout in worker.earlier_projects assert not rule.owns_present_checkout(worker, source, RepositoryReads()) diff --git a/tests/test_project.py b/tests/test_project.py index ed79f44c3e..5a8daa3027 100644 --- a/tests/test_project.py +++ b/tests/test_project.py @@ -17,6 +17,7 @@ from agent_comms.comms import wire from agent_comms.acp_extension import CoordinationChangedUpdate, decode_updates from agent_comms.native_arguments import GivenArguments, GivenBinary +from agent_comms.repo import Checkout async def test_project_update_is_published_and_old_saved_cwd_can_resume( @@ -37,13 +38,13 @@ async def publish_update(self, **kwargs): return True agent.on_connect(Client()) - comms.threads.set_project(session, str(new)) + comms.threads.set_project(comms.registry.require(session).incarnation, str(new)) await agent.sessions.sync_identity(session) assert next( update.worktree for update in decode_updates(updates[-1].field_meta) if isinstance(update, CoordinationChangedUpdate) - ) == str(new) + ) == Checkout(new) await agent.shutdown() resumed = canonical_agent(comms, agent_bin=GivenBinary("/bin/echo"), agent_args=GivenArguments(())) try: @@ -52,7 +53,7 @@ async def publish_update(self, **kwargs): update.worktree for update in decode_updates(result.field_meta) if isinstance(update, CoordinationChangedUpdate) - ) == str(new) + ) == Checkout(new) with pytest.raises(RequestError): await resumed.load_session(str(tmp_path / "unrelated"), session) assert len(comms.registry.all_threads()) == 1 @@ -88,13 +89,13 @@ async def publish_update(self, session_id, update): update.worktree for update in decode_updates(await proxy.subscribe()) if isinstance(update, CoordinationChangedUpdate) - ) == str(old) - comms.threads.set_project(session, str(new)) + ) == Checkout(old) + comms.threads.set_project(comms.registry.require(session).incarnation, str(new)) await owner.sessions.sync_identity(session) async with asyncio.timeout(2): while not all( any( - update.worktree == str(new) + update.worktree == Checkout(new) for value in values for update in decode_updates(value.get("_meta")) if isinstance(update, CoordinationChangedUpdate) diff --git a/tests/test_project_resume.py b/tests/test_project_resume.py index 75ac97cd5b..07f667bc25 100644 --- a/tests/test_project_resume.py +++ b/tests/test_project_resume.py @@ -31,14 +31,14 @@ def declared_owner(tmp_path): def test_project_change_while_idle_owes_no_turn(tmp_path): comms, _old, new = declared_owner(tmp_path) - assert comms.threads.set_project("owner", str(new)).changed + assert comms.threads.set_project(comms.registry.require("owner").incarnation, str(new)).changed assert not comms.registry.require("owner").project_resume_due def test_project_change_during_turn_is_recorded_until_the_next_turn_starts(tmp_path): comms, old, new = declared_owner(tmp_path) turn = comms.agents.begin_turn("owner", TurnId('1056fa0e005902e019b8d1d4fd75d3cb')) - comms.threads.set_project("owner", str(new)) + comms.threads.set_project(comms.registry.require("owner").incarnation, str(new)) assert comms.registry.require("owner").project_resume_due assert comms.agents.finish_turn(turn.turn_lease) # The turn ended and nothing ran afterwards: the registry still says so, @@ -55,7 +55,7 @@ def test_project_change_during_turn_is_recorded_until_the_next_turn_starts(tmp_p def test_cancelled_turn_withdraws_the_recorded_resume(tmp_path): comms, _old, new = declared_owner(tmp_path) turn = comms.agents.begin_turn("owner", TurnId('1056fa0e005902e019b8d1d4fd75d3cb')) - comms.threads.set_project("owner", str(new)) + comms.threads.set_project(comms.registry.require("owner").incarnation, str(new)) comms.agents.transition_turn(turn.turn_lease, CancellingPhase()) assert comms.agents.finish_turn(turn.turn_lease) current = comms.registry.require("owner") @@ -80,7 +80,7 @@ def reply(message): ),))) def change_project_and_send_again(): - changed = comms.threads.set_project(session, str(second)) + changed = comms.threads.set_project(comms.registry.require(session).incarnation, str(second)) sent.append(comms.messaging.send("sender", session, "Second message").message) return changed diff --git a/tests/test_project_source_identity.py b/tests/test_project_source_identity.py new file mode 100644 index 0000000000..db7f7cf43b --- /dev/null +++ b/tests/test_project_source_identity.py @@ -0,0 +1,55 @@ +"""Project selection retains the caller's original registry identity.""" +from dataclasses import replace + +import pytest + +from agent_comms.comms import wire +from agent_comms.errors import RelationViolationError +from agent_comms.repo import Checkout +from agent_comms.thread_text import GivenThreadText +from agent_comms.threads import Thread + + +@pytest.fixture +def project_source(tmp_path): + current, destination = tmp_path / "current", tmp_path / "destination" + current.mkdir() + destination.mkdir() + comms = wire(tmp_path / "wire") + original = comms.registry.declare(Thread("source", frozenset(), Checkout(current))) + return comms, original, destination + + +def test_captured_project_target_follows_its_original_rename(project_source): + comms, original, destination = project_source + comms.registry.rename(original.name, "renamed") + changed = comms.threads.set_project(original.incarnation, str(destination)) + current = comms.registry.require("renamed") + assert changed.thread == current.name == "renamed" + assert current.incarnation == original.incarnation + assert current.worktree == Checkout(destination) + assert current.earlier_projects == (original.worktree,) + + +def test_recreated_name_cannot_receive_the_original_project_change(project_source): + comms, original, destination = project_source + with comms.registry.store.editing() as edit: + edit.document.remove(original.name) + edit.commit() + replacement = comms.registry.declare(Thread( + original.name, frozenset(), original.worktree)) + assert replacement.incarnation != original.incarnation + with pytest.raises(RelationViolationError): + comms.threads.set_project(original.incarnation, str(destination)) + assert comms.registry.require(replacement.name) == replacement + assert not comms.registry.require(replacement.name).earlier_projects + + +def test_same_birth_metadata_does_not_retire_a_project_target(project_source): + comms, original, destination = project_source + titled = comms.registry.declare(replace(original, title=GivenThreadText("new title"))) + changed = comms.threads.set_project(original.incarnation, str(destination)) + current = comms.registry.require(original.name) + assert changed.changed and current.worktree == Checkout(destination) + assert current.title == titled.title + assert current.incarnation == original.incarnation diff --git a/tests/test_registration_inheritance.py b/tests/test_registration_inheritance.py index 3befe58573..e192f8b105 100644 --- a/tests/test_registration_inheritance.py +++ b/tests/test_registration_inheritance.py @@ -93,9 +93,9 @@ def test_project_round_trip_excludes_current_project_from_history(tmp_path): first.mkdir() second.mkdir() comms = wire(tmp_path / 'wire') - comms.registry.declare(Thread('owner', frozenset(), Checkout(first))) - comms.threads.set_project('owner', str(second)) - comms.threads.set_project('owner', str(first)) + original = comms.registry.declare(Thread('owner', frozenset(), Checkout(first))) + comms.threads.set_project(original.incarnation, str(second)) + comms.threads.set_project(original.incarnation, str(first)) current = comms.registry.require('owner') assert current.worktree == Checkout(first) assert current.earlier_projects == (Checkout(second),) @@ -109,7 +109,7 @@ def test_project_change_retains_executing_source_but_refuses_another_input(tmp_p comms.registry.declare(Thread('owner', frozenset(), Checkout(old), process_identity=ProcessIdentity.capture(os.getpid()))) original = comms.agents.begin_turn('owner', TurnId('ab87f599dc0a50a56b7c65de71c39426')) try: - comms.threads.set_project('owner', str(new)) + comms.threads.set_project(original.thread.incarnation, str(new)) with pytest.raises(StaleFence): original.require_snapshot(comms.registry.snapshot(), 'New input refused') saved = str(old / 'native.jsonl') diff --git a/tests/test_session_load_plan.py b/tests/test_session_load_plan.py index 77c79ad95f..44c04ae30c 100644 --- a/tests/test_session_load_plan.py +++ b/tests/test_session_load_plan.py @@ -1,4 +1,6 @@ -"""Public type names may change; the existing ACP metadata must not.""" +"""Original ACP load witnesses retain their wire contract and owner semantics.""" + +import pytest from agent_comms.child_process import ProcessIdentity from agent_comms.nominal.field_codec import FieldCodec @@ -9,7 +11,7 @@ SessionLoadPlan, ) from agent_comms.thread_identity import OwnerIdentity, RecordedThreadIncarnation -from agent_comms.thread_presentation import LiveThreadOwnerBinding +from agent_comms.thread_presentation import LiveThreadOwnerBinding, UnavailableThreadOwnerBinding def test_default_load_metadata_keeps_its_original_shape(): @@ -38,3 +40,46 @@ def test_witnessed_load_and_failure_metadata_keep_original_tags(): "agentCommsLoadFailure": {"kind": "failed", "binding": FieldCodec.encode(binding)} } assert FailedSessionLoadPlan.from_failure(error) == failed + + +@pytest.mark.parametrize("plan_type", (ExistingSessionLoadPlan, FailedSessionLoadPlan)) +async def test_only_a_superseding_live_owner_supplies_an_existing_load(plan_type): + original = LiveThreadOwnerBinding( + OwnerIdentity(RecordedThreadIncarnation("original", 123.0), 2), + ProcessIdentity(1, 1), + ) + previous = plan_type(original) + loads: list[SessionLoadPlan] = [] + + async def reload(plan: SessionLoadPlan) -> None: + loads.append(plan) + + await UnavailableThreadOwnerBinding().reload_if_superseded(previous, reload) + await original.reload_if_superseded(previous, reload) + await LiveThreadOwnerBinding( + OwnerIdentity(original.owner.incarnation, 1), ProcessIdentity(2, 2), + ).reload_if_superseded(previous, reload) + await LiveThreadOwnerBinding( + OwnerIdentity(RecordedThreadIncarnation("replacement", 124.0), 3), + ProcessIdentity(2, 2), + ).reload_if_superseded(previous, reload) + assert not loads + renamed = original.owner.incarnation.renamed("renamed") + successor = LiveThreadOwnerBinding(OwnerIdentity(renamed, 3), ProcessIdentity(2, 2)) + await successor.reload_if_superseded(previous, reload) + assert loads == [ExistingSessionLoadPlan(successor)] + assert previous.binding == original + + +async def test_an_ensuring_load_is_not_an_old_owner_witness(): + binding = LiveThreadOwnerBinding( + OwnerIdentity(RecordedThreadIncarnation("source", 123.0), 2), + ProcessIdentity(1, 1), + ) + loads: list[SessionLoadPlan] = [] + + async def reload(plan: SessionLoadPlan) -> None: + loads.append(plan) + + await binding.reload_if_superseded(EnsuringSessionLoadPlan(), reload) + assert not loads From f84b4cbfa7f07b8829e237bea8e1d7ddae835ec3 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Sun, 1 Nov 2026 22:46:12 -0500 Subject: [PATCH 02/10] Let witnessed load plans own reload construction --- src/agent_comms/session_load.py | 14 ++++++++++++++ src/agent_comms/thread_presentation.py | 5 +---- 2 files changed, 15 insertions(+), 4 deletions(-) diff --git a/src/agent_comms/session_load.py b/src/agent_comms/session_load.py index 50b5a215eb..37d13fa780 100644 --- a/src/agent_comms/session_load.py +++ b/src/agent_comms/session_load.py @@ -1,5 +1,6 @@ """The declared session type checks owner permission when handling an ACP load.""" from abc import abstractmethod +from collections.abc import Awaitable, Callable from functools import partial from .coordinator import Coordination from dataclasses import dataclass @@ -35,6 +36,12 @@ def at_response(cls, metadata: object) -> 'SessionLoadPlan': def superseded_by(self, binding: ThreadOwnerBinding) -> bool: return False + async def reload_if_superseded( + self, binding: LiveThreadOwnerBinding, + reload: Callable[['SessionLoadPlan'], Awaitable[None]], + ) -> None: + """A plan without an owner witness cannot request an observed reload.""" + @abstractmethod async def resolve(self, lifecycle: "AttachedSessionLifecycle", thread: "Thread") -> "Thread": ... @@ -58,6 +65,13 @@ class WitnessedSessionLoadPlan(SessionLoadPlan): def superseded_by(self, binding: ThreadOwnerBinding) -> bool: return self.binding.superseded_by(binding) + async def reload_if_superseded( + self, binding: LiveThreadOwnerBinding, + reload: Callable[[SessionLoadPlan], Awaitable[None]], + ) -> None: + if self.superseded_by(binding): + await reload(ExistingSessionLoadPlan(binding)) + @dataclass(frozen=True, slots=True) class ExistingSessionLoadPlan(WitnessedSessionLoadPlan): diff --git a/src/agent_comms/thread_presentation.py b/src/agent_comms/thread_presentation.py index 3ae429e7bf..d40e64d3e0 100644 --- a/src/agent_comms/thread_presentation.py +++ b/src/agent_comms/thread_presentation.py @@ -195,10 +195,7 @@ def superseded_by(self, replacement: ThreadOwnerBinding) -> bool: async def reload_if_superseded( self, previous: SessionLoadPlan, reload: Callable[[SessionLoadPlan], Awaitable[None]], ) -> None: - from .session_load import ExistingSessionLoadPlan - - if previous.superseded_by(self): - await reload(ExistingSessionLoadPlan(self)) + await previous.reload_if_superseded(self, reload) def replaces(self, original: LiveThreadOwnerBinding) -> bool: # Both operands are the ORIGINAL registry owner/process witnesses. From e910a54829ae163428cd722b1150e8ff019b52e1 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Sun, 1 Nov 2026 23:18:20 -0500 Subject: [PATCH 03/10] Separate transcript history from ACP publication identity --- src/agent_comms/acp.py | 4 ++-- src/agent_comms/runtime_requests.py | 4 +++- src/agent_comms/session_lifecycle.py | 8 ++++++-- src/agent_comms/transcript_updates.py | 19 ++++++++++++------- tests/test_acp.py | 20 +++++++++++++++++--- tests/test_runtime.py | 3 ++- tests/test_transcript_input_display.py | 13 ++++++++++--- 7 files changed, 52 insertions(+), 19 deletions(-) diff --git a/src/agent_comms/acp.py b/src/agent_comms/acp.py index 30a39596cc..8f5819dec7 100644 --- a/src/agent_comms/acp.py +++ b/src/agent_comms/acp.py @@ -76,7 +76,7 @@ from .coordination_database import CoordinationStore from .cursor_owner import CursorOwner from .cursor_publication import CursorPublication -from .nominal.field_codec import AcpSessionId, ClaimId, FieldCodec, RootId +from .nominal.field_codec import AcpSessionId, AcpSessionIdText, ClaimId, FieldCodec, RootId from .image_inputs import prompt_images from .input_drain import InputDrain from .config_options import ConfigOptions @@ -192,7 +192,7 @@ class SdkSessionConnection(AgentSideConnection, Controller, SessionUpdateSink): """The SDK connection and controller publish one Core acceptance answer.""" async def publish_update(self, *, session_id: AcpSessionId, update: BaseModel) -> bool: - await self.session_update(session_id=session_id, update=update) + await self.session_update(session_id=AcpSessionIdText.encode(session_id), update=update) return True diff --git a/src/agent_comms/runtime_requests.py b/src/agent_comms/runtime_requests.py index 1d76386664..344f39252b 100644 --- a/src/agent_comms/runtime_requests.py +++ b/src/agent_comms/runtime_requests.py @@ -161,7 +161,9 @@ async def observing(self, ctx: AttachedRuntimeRequestContext) -> AsyncIterator[N agent = ctx.server.agent ctx.server.clients.setdefault(ctx.session, set()).add(ctx.client) await agent.emit_session_identity(ctx.session, ctx.name, client=ctx.client) - async with agent.sessions.transcript.subscribe(self.thread, client=ctx.client): + async with agent.sessions.transcript.subscribe( + ctx.name, session_id=ctx.session, client=ctx.client, + ): await agent.turns.publish_turn_state( AcpEventConsumer(agent, ctx.session, ctx.client) ) diff --git a/src/agent_comms/session_lifecycle.py b/src/agent_comms/session_lifecycle.py index 3b6522c249..8860c90fc6 100644 --- a/src/agent_comms/session_lifecycle.py +++ b/src/agent_comms/session_lifecycle.py @@ -249,7 +249,9 @@ async def load_session(self, request: LoadSessionRequest) -> LoadSessionResponse if not thread.process_identity.runs_here(): return await self.attach_owner(thread, session_id) await self._bind_loaded(thread, session_id) - await self.transcript.replay(session_id, self.runtime) + await self.transcript.replay( + thread.name, session_id=session_id, client=self.runtime, + ) self.effects.inputs.ensure_live_drain(session_id) response = LoadSessionResponse( field_meta=await self.metadata(thread.name, session_id=session_id), @@ -355,7 +357,9 @@ async def attach_owner( self.proxies[session_id] = proxy except IncompatibleOwnerError as error: # History is Core's own store; only owner requests need the owner's Core. - await self.transcript.replay(session_id, self.runtime) + await self.transcript.replay( + thread.name, session_id=session_id, client=self.runtime, + ) self.proxies[session_id] = error.owner return LoadSessionResponse(field_meta={ **await self.metadata(thread.name, session_id=session_id), diff --git a/src/agent_comms/transcript_updates.py b/src/agent_comms/transcript_updates.py index 4c0f7029d4..edbf04f49a 100644 --- a/src/agent_comms/transcript_updates.py +++ b/src/agent_comms/transcript_updates.py @@ -17,6 +17,7 @@ ) from .coordinator import Coordination from .nominal.declared_family import DeclaredFamily +from .nominal.field_codec import AcpSessionId from .order_tools import TranscriptPageRead from .turn_lease import TurnState from .ui_model.observation import ObservationProcess, ReadTranscript, SubscribeTranscript @@ -31,7 +32,7 @@ class TranscriptUpdate(DeclaredFamily, affix="TranscriptUpdate"): @abstractmethod async def publish( - self, session_id: str, client: SessionUpdateSink, + self, session_id: AcpSessionId, client: SessionUpdateSink, ) -> None: ... @@ -40,7 +41,7 @@ class TurnTranscriptUpdate(TranscriptUpdate): state: TurnState async def publish( - self, session_id: str, client: SessionUpdateSink, + self, session_id: AcpSessionId, client: SessionUpdateSink, ) -> None: await client.publish_update( session_id=session_id, @@ -74,25 +75,29 @@ async def observing(self) -> AsyncIterator[ObservationProcess]: if not self._subscriptions: await reads.close() - async def replay(self, session_id: str, client: SessionUpdateSink) -> None: + async def replay( + self, thread: str, *, session_id: AcpSessionId, client: SessionUpdateSink, + ) -> None: async with self.observing() as reads: snapshot = await reads.read( - ReadTranscript(str(self.comms.root), TranscriptPageRead(thread=session_id)), + ReadTranscript(str(self.comms.root), TranscriptPageRead(thread=thread)), ) await self._publish(session_id, snapshot, client) @asynccontextmanager - async def subscribe(self, session_id: str, client: SessionUpdateSink) -> AsyncIterator[None]: + async def subscribe( + self, thread: str, *, session_id: AcpSessionId, client: SessionUpdateSink, + ) -> AsyncIterator[None]: async with self.observing() as reads: snapshot = await reads.read( - SubscribeTranscript(str(self.comms.root), TranscriptPageRead(thread=session_id)), + SubscribeTranscript(str(self.comms.root), TranscriptPageRead(thread=thread)), ) await self._publish(session_id, snapshot, client) yield async def _publish( self, - session_id: str, + session_id: AcpSessionId, snapshot: TranscriptSnapshotUpdate, destination: SessionUpdateSink, ) -> None: diff --git a/tests/test_acp.py b/tests/test_acp.py index 228ac22f3c..ff791a5cf7 100644 --- a/tests/test_acp.py +++ b/tests/test_acp.py @@ -52,9 +52,11 @@ class RecordingClient(SessionUpdateSink): def __init__(self): self.updates = [] + self.destinations = [] async def publish_update(self, session_id=None, update=None, **kwargs): self.updates.append(update) + self.destinations.append(session_id) return True @@ -178,8 +180,9 @@ async def test_compaction_details_preserve_markdown(self, tmp_path): assert "\x1b" not in backend.compaction_summary("\x1b[2J\nSafe") @pytest.mark.parametrize("renamed", [False, True]) + @pytest.mark.parametrize("subscribed", [False, True]) async def test_snapshot_capability_replays_one_bounded_update( - self, tmp_path, renamed, native_runtime_package + self, tmp_path, renamed, subscribed, native_runtime_package ): agent = self._agent(tmp_path) await agent.initialize(1) @@ -206,11 +209,22 @@ async def test_snapshot_capability_replays_one_bounded_update( agent._comms.threads.rename_managed_thread("proj", "renamed", owner=ProcessIdentity.capture(os.getpid())) client = RecordingClient() sent = client.updates + destination = AcpSessionId("snapshot-destination") + thread = agent._comms.registry.require("proj") try: - await agent.sessions.transcript.replay("proj", client=client) + if subscribed: + async with agent.sessions.transcript.subscribe( + thread.name, session_id=destination, client=client, + ): + assert len(sent) == 1 + else: + await agent.sessions.transcript.replay( + thread.name, session_id=destination, client=client, + ) + assert client.destinations == [destination] assert len(sent) == 1 update = facts(sent[0].field_meta, TranscriptSnapshotUpdate)[0] - assert update.identity.requested_name == "proj" + assert update.identity.requested_name == thread.name assert update.identity.thread.incarnation == agent._comms.registry.require("proj").incarnation page = update.page events = page.events diff --git a/tests/test_runtime.py b/tests/test_runtime.py index 0a658750ec..c9acc0def6 100644 --- a/tests/test_runtime.py +++ b/tests/test_runtime.py @@ -250,7 +250,8 @@ async def publish_update(self, session_id, update): assert identity_position < replay_position assert len(snapshots) == 1 session, snapshot = snapshots[0] - assert session == requested == snapshot.identity.requested_name + assert session == requested + assert snapshot.identity.requested_name == "renamed" assert snapshot.identity.thread.incarnation == comms.registry.require("renamed").incarnation assert [event.text for event in snapshot.page.events] == ["An original saved answer"] finally: diff --git a/tests/test_transcript_input_display.py b/tests/test_transcript_input_display.py index 671f0e1d2e..b7d04f4d5f 100644 --- a/tests/test_transcript_input_display.py +++ b/tests/test_transcript_input_display.py @@ -14,6 +14,7 @@ from agent_comms.comms import wire from conftest import installed_comms from agent_comms.native_entries import TranscriptProjection +from agent_comms.nominal.field_codec import AcpSessionId from agent_comms.pi_payloads import PiMessage, UserMessage from agent_comms.threads import Thread from agent_comms.transcript_routes import DisplayText, InputDisplay, NoDisplayText, SentInputText @@ -109,14 +110,20 @@ async def test_acp_saved_transcript_replay_hides_only_owned_internal_input(tmp_p comms.transcripts.routes.record_input_display(SECOND_INPUT_ID, DisplayText("test2")) agent = CommsAgent(wire(comms.root)) updates = [] + destinations = [] + destination = AcpSessionId("input-display-destination") class Client(SessionUpdateSink): - async def publish_update(self, **kwargs): - updates.append(kwargs["update"]) + async def publish_update(self, *, session_id, update): + destinations.append(session_id) + updates.append(update) return True try: - await agent.sessions.transcript.replay("worker", client=Client()) + await agent.sessions.transcript.replay( + "worker", session_id=destination, client=Client(), + ) + assert destinations == [destination] facts = decode_updates(updates[0].field_meta) snapshot, = (fact for fact in facts if isinstance(fact, TranscriptSnapshotUpdate)) events = snapshot.page.events From 992855d7dbd261841c3a652b2033cfc896f5b318 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Mon, 2 Nov 2026 00:03:52 -0500 Subject: [PATCH 04/10] Decode saved input fixture IDs at their producer --- tests/test_transcript_input_display.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/tests/test_transcript_input_display.py b/tests/test_transcript_input_display.py index b7d04f4d5f..64e5506565 100644 --- a/tests/test_transcript_input_display.py +++ b/tests/test_transcript_input_display.py @@ -17,6 +17,7 @@ from agent_comms.nominal.field_codec import AcpSessionId from agent_comms.pi_payloads import PiMessage, UserMessage from agent_comms.threads import Thread +from agent_comms.turn_item import FreshNativeIdText from agent_comms.transcript_routes import DisplayText, InputDisplay, NoDisplayText, SentInputText from bus_record_fixture import FIRST_INPUT_ID, SECOND_INPUT_ID, THIRD_INPUT_ID @@ -32,7 +33,8 @@ def saved_row(native_id, text, *, images=False, role="user"): if images: content.append({"type": "image", "mimeType": "image/png", "data": "secret-image"}) return transcript_records.message( - {"role": role, "inputId": str(native_id), "content": content}, id=native_id[:8] + {"role": role, "inputId": str(native_id), "content": content}, + id=FreshNativeIdText.decode(native_id[:8]), ) From 2ddaa22f937f0184abd089c220d986973d4cc82f Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Mon, 2 Nov 2026 00:55:37 -0500 Subject: [PATCH 05/10] Use native deferred return annotation on failed session load --- src/agent_comms/session_load.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/agent_comms/session_load.py b/src/agent_comms/session_load.py index 37d13fa780..aca3eecb83 100644 --- a/src/agent_comms/session_load.py +++ b/src/agent_comms/session_load.py @@ -94,7 +94,7 @@ class FailedSessionLoadPlan(WitnessedSessionLoadPlan): """The original failed load command's witness; never an input permission.""" @classmethod - def from_failure(cls, data: object) -> 'FailedSessionLoadPlan | None': + def from_failure(cls, data: object) -> FailedSessionLoadPlan | None: if not isinstance(data, dict): return None witness = data.get("agentCommsLoadFailure") From 97e166e201e3ea787775f2c27f8e2029c52b5841 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Mon, 2 Nov 2026 01:37:20 -0500 Subject: [PATCH 06/10] Preserve composed project fixtures through submission landing --- tests/test_project.py | 13 ++++++------- tests/test_project_resume.py | 2 +- 2 files changed, 7 insertions(+), 8 deletions(-) diff --git a/tests/test_project.py b/tests/test_project.py index 5a8daa3027..f60d9c9274 100644 --- a/tests/test_project.py +++ b/tests/test_project.py @@ -16,7 +16,6 @@ from delivery_owner_fixture import canonical_agent from agent_comms.comms import wire from agent_comms.acp_extension import CoordinationChangedUpdate, decode_updates -from agent_comms.native_arguments import GivenArguments, GivenBinary from agent_comms.repo import Checkout @@ -28,8 +27,8 @@ async def test_project_update_is_published_and_old_saved_cwd_can_resume( old.mkdir() new.mkdir() comms = wire(tmp_path / "wire") - agent = canonical_agent(comms, agent_bin=GivenBinary("/bin/echo"), agent_args=GivenArguments(())) - session = (await agent.new_session(str(old))).session_id + agent = canonical_agent(comms) + session = FieldCodec.decode(AcpSessionId, (await agent.new_session(str(old))).session_id) updates = [] class Client: @@ -46,16 +45,16 @@ async def publish_update(self, **kwargs): if isinstance(update, CoordinationChangedUpdate) ) == Checkout(new) await agent.shutdown() - resumed = canonical_agent(comms, agent_bin=GivenBinary("/bin/echo"), agent_args=GivenArguments(())) + resumed = canonical_agent(comms) try: - result = await resumed.load_session(str(old), session) + result = await resumed.load_session(str(old), FieldCodec.encode(session)) assert next( update.worktree for update in decode_updates(result.field_meta) if isinstance(update, CoordinationChangedUpdate) ) == Checkout(new) with pytest.raises(RequestError): - await resumed.load_session(str(tmp_path / "unrelated"), session) + await resumed.load_session(str(tmp_path / "unrelated"), FieldCodec.encode(session)) assert len(comms.registry.all_threads()) == 1 finally: await resumed.shutdown() @@ -67,7 +66,7 @@ async def test_project_changes_reach_all_subscribed_clients(tmp_path, native_run old.mkdir() new.mkdir() comms = wire(tmp_path / "wire") - owner = canonical_agent(comms, agent_bin=GivenBinary("/bin/echo"), agent_args=GivenArguments(()), runtime_enabled=True) + owner = canonical_agent(comms, runtime_enabled=True) session = FieldCodec.decode(AcpSessionId, (await owner.new_session(str(old))).session_id) proxies, updates = [], [[], []] diff --git a/tests/test_project_resume.py b/tests/test_project_resume.py index 07f667bc25..c3dbcaa365 100644 --- a/tests/test_project_resume.py +++ b/tests/test_project_resume.py @@ -75,7 +75,7 @@ async def test_message_after_project_change_starts_a_turn_in_the_new_project(tmp def reply(message): return json.dumps(FieldCodec.encode((Message( - session, message.exact_reply_target, f"Reply {len(sent)}", + agent.sessions.require(session), message.exact_reply_target, f"Reply {len(sent)}", InfoKind, timestamp=0, ),))) From abc44518da9a98331d508b05b606a7c0960d4abf Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Mon, 2 Nov 2026 06:21:57 -0500 Subject: [PATCH 07/10] Resolve load-plan callback types through their declaring modules --- src/agent_comms/session_load.py | 22 +++++++++++----------- src/agent_comms/thread_presentation.py | 12 +++++++++--- 2 files changed, 20 insertions(+), 14 deletions(-) diff --git a/src/agent_comms/session_load.py b/src/agent_comms/session_load.py index aca3eecb83..8f5f8403ce 100644 --- a/src/agent_comms/session_load.py +++ b/src/agent_comms/session_load.py @@ -2,14 +2,14 @@ from abc import abstractmethod from collections.abc import Awaitable, Callable from functools import partial -from .coordinator import Coordination +import agent_comms.coordinator as coordinator from dataclasses import dataclass from typing import TYPE_CHECKING from .nominal.declared_family import DeclaredFamily from .errors import RelationViolationError from .nominal.field_codec import FieldCodec -from .thread_presentation import LiveThreadOwnerBinding, ThreadOwnerBinding +import agent_comms.thread_presentation as thread_presentation from .native_arguments import GivenArguments if TYPE_CHECKING: @@ -33,12 +33,12 @@ def at_response(cls, metadata: object) -> 'SessionLoadPlan': return (EnsuringSessionLoadPlan() if witness is None else FieldCodec.decode(ExistingSessionLoadPlan, witness)) - def superseded_by(self, binding: ThreadOwnerBinding) -> bool: + def superseded_by(self, binding: thread_presentation.ThreadOwnerBinding) -> bool: return False async def reload_if_superseded( - self, binding: LiveThreadOwnerBinding, - reload: Callable[['SessionLoadPlan'], Awaitable[None]], + self, binding: thread_presentation.LiveThreadOwnerBinding, + reload: Callable[[SessionLoadPlan], Awaitable[None]], ) -> None: """A plan without an owner witness cannot request an observed reload.""" @@ -49,7 +49,7 @@ async def resolve(self, lifecycle: "AttachedSessionLifecycle", thread: "Thread") @dataclass(frozen=True, slots=True) class EnsuringSessionLoadPlan(SessionLoadPlan): async def resolve(self, lifecycle: "AttachedSessionLifecycle", thread: "Thread") -> "Thread": - return await Coordination.run_worker(partial( + return await coordinator.Coordination.run_worker(partial( lifecycle.comms.owners.ensure_owner, thread.name, agent_bin=lifecycle.effects.inputs.configuration.launch.binary, agent_args=GivenArguments(lifecycle.effects.inputs.configuration.agent_args.argv), @@ -60,13 +60,13 @@ async def resolve(self, lifecycle: "AttachedSessionLifecycle", thread: "Thread") class WitnessedSessionLoadPlan(SessionLoadPlan): """Original load command/response evidence, never current registry state.""" - binding: ThreadOwnerBinding + binding: thread_presentation.ThreadOwnerBinding - def superseded_by(self, binding: ThreadOwnerBinding) -> bool: + def superseded_by(self, binding: thread_presentation.ThreadOwnerBinding) -> bool: return self.binding.superseded_by(binding) async def reload_if_superseded( - self, binding: LiveThreadOwnerBinding, + self, binding: thread_presentation.LiveThreadOwnerBinding, reload: Callable[[SessionLoadPlan], Awaitable[None]], ) -> None: if self.superseded_by(binding): @@ -75,10 +75,10 @@ async def reload_if_superseded( @dataclass(frozen=True, slots=True) class ExistingSessionLoadPlan(WitnessedSessionLoadPlan): - binding: LiveThreadOwnerBinding + binding: thread_presentation.LiveThreadOwnerBinding async def resolve(self, lifecycle: "AttachedSessionLifecycle", thread: "Thread") -> "Thread": - entry = (await Coordination.run_worker(partial( + entry = (await coordinator.Coordination.run_worker(partial( lifecycle.comms.registry.entry, thread.name, ))).entry if thread.incarnation != self.binding.owner.incarnation: diff --git a/src/agent_comms/thread_presentation.py b/src/agent_comms/thread_presentation.py index d40e64d3e0..9be71ec129 100644 --- a/src/agent_comms/thread_presentation.py +++ b/src/agent_comms/thread_presentation.py @@ -16,7 +16,6 @@ from .message_reference import MessageReference if TYPE_CHECKING: - from .session_load import SessionLoadPlan from .presentation import MessageNotification from .transcripts import AcquiredThreadPresentation, TranscriptReadIdentity from agent_comms.registry_document import RegistrySnapshot @@ -151,7 +150,8 @@ def superseded_by(self, replacement: ThreadOwnerBinding) -> bool: return False async def reload_if_superseded( - self, previous: SessionLoadPlan, reload: Callable[[SessionLoadPlan], Awaitable[None]], + self, previous: session_load.SessionLoadPlan, + reload: Callable[[session_load.SessionLoadPlan], Awaitable[None]], ) -> None: """An unavailable owner cannot supply a new existing-session load.""" @@ -193,7 +193,8 @@ def superseded_by(self, replacement: ThreadOwnerBinding) -> bool: return replacement.replaces(self) async def reload_if_superseded( - self, previous: SessionLoadPlan, reload: Callable[[SessionLoadPlan], Awaitable[None]], + self, previous: session_load.SessionLoadPlan, + reload: Callable[[session_load.SessionLoadPlan], Awaitable[None]], ) -> None: await previous.reload_if_superseded(self, reload) @@ -303,3 +304,8 @@ def read_sources(self, transcripts: Transcripts, snapshot: RegistrySnapshot, name: str, sent_source_limit: int) -> ThreadPresentation: return self + +# Session-load annotations use the module namespace. Publish these declarations +# before loading its coordinator dependencies, which consume our status classes. +import agent_comms.session_load as session_load + From 4f7e2c138d2936ed75d3b6652732695a87f964dd Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Mon, 2 Nov 2026 06:39:42 -0500 Subject: [PATCH 08/10] Factor joined coordination work out of store bootstrap --- src/agent_comms/coordination_work.py | 28 ++++++++++++++++++++++++++++ src/agent_comms/coordinator.py | 25 +++++-------------------- src/agent_comms/session_load.py | 8 ++++---- 3 files changed, 37 insertions(+), 24 deletions(-) create mode 100644 src/agent_comms/coordination_work.py diff --git a/src/agent_comms/coordination_work.py b/src/agent_comms/coordination_work.py new file mode 100644 index 0000000000..369e833d51 --- /dev/null +++ b/src/agent_comms/coordination_work.py @@ -0,0 +1,28 @@ +"""Joined blocking work without opening or importing coordination stores.""" + +from collections.abc import Callable +from contextvars import Context +from functools import partial + +from .child_process import AsyncWork +from .maintenance_barrier import MaintenanceHolder +from .native_build_inputs import GitLaunch +from .store_files import StoreLock + + +class CoordinationWork: + """The shared resource context for blocking store and registry operations.""" + + @staticmethod + def worker_context() -> Context: + return GitLaunch.in_context(MaintenanceHolder.in_context(StoreLock.worker_context())) + + @staticmethod + async def run_worker[Result](operation: Callable[[], Result]) -> Result: + """Wait for blocking work before its caller releases the enclosing resources. + + The callback owns and closes every borrowed resource it opens. This + scheduling seam also supports registry-only acquisition/retirement; + those operations must not depend on opening a coordinator connection. + """ + return await AsyncWork.run_async(partial(CoordinationWork.worker_context().run, operation)) diff --git a/src/agent_comms/coordinator.py b/src/agent_comms/coordinator.py index 39b307b4f3..d4a619aa49 100644 --- a/src/agent_comms/coordinator.py +++ b/src/agent_comms/coordinator.py @@ -8,15 +8,15 @@ from collections.abc import AsyncIterator, Callable from concurrent.futures import ThreadPoolExecutor from contextlib import AbstractContextManager, ExitStack, asynccontextmanager -from contextvars import Context, ContextVar +from contextvars import ContextVar from functools import partial from pathlib import Path from typing import TYPE_CHECKING, ClassVar, Final, Self, TypeVar from .assignment_store import AssignmentStore from .attempt_store import AttemptStore -from .child_process import AsyncWork, join_retirement -from .native_build_inputs import GitLaunch +from .child_process import join_retirement +from .coordination_work import CoordinationWork from .coordination_session import ( CoordinationSession, NativeCoordinationSession, @@ -24,9 +24,8 @@ system_clock, ) from .execution_store import ExecutionStore -from .maintenance_barrier import MaintenanceHolder from .participant_store import ParticipantStore -from .store_files import DEFAULT_STORE_CONTENTION, StoreContention, StoreLock +from .store_files import DEFAULT_STORE_CONTENTION, StoreContention from .working_memory_annotations import WorkingMemoryAnnotations if TYPE_CHECKING: @@ -96,7 +95,7 @@ async def run_read( STANDALONE_ROUTE: Final = StandaloneRoute() -class Coordination(ExecutionStore, AttemptStore): +class Coordination(CoordinationWork, ExecutionStore, AttemptStore): watch_period: ClassVar[float] = 60 session_type = NativeCoordinationSession _worker: ContextVar[CoordinationRoute] = ContextVar( @@ -115,20 +114,6 @@ async def worker(cls, path: str | Path) -> AsyncIterator[CoordinationWorker]: finally: cls._worker.reset(token) - @staticmethod - def worker_context() -> Context: - return GitLaunch.in_context(MaintenanceHolder.in_context(StoreLock.worker_context())) - - @staticmethod - async def run_worker(operation: Callable[[], Result]) -> Result: - """Wait for blocking work before its caller releases the enclosing resources. - - The callback owns and closes every borrowed resource it opens. This - scheduling seam also supports registry-only acquisition/retirement; - those operations must not depend on opening a coordinator connection. - """ - return await AsyncWork.run_async(partial(Coordination.worker_context().run, operation)) - @classmethod async def watch_worker( cls, watch: Watch, opened_wire: Callable[[], AbstractContextManager[Comms]], diff --git a/src/agent_comms/session_load.py b/src/agent_comms/session_load.py index 8f5f8403ce..906d1b48ea 100644 --- a/src/agent_comms/session_load.py +++ b/src/agent_comms/session_load.py @@ -2,7 +2,7 @@ from abc import abstractmethod from collections.abc import Awaitable, Callable from functools import partial -import agent_comms.coordinator as coordinator +from .coordination_work import CoordinationWork from dataclasses import dataclass from typing import TYPE_CHECKING @@ -17,7 +17,7 @@ from .threads import Thread -class SessionLoadPlan(DeclaredFamily, affix='SessionLoadPlan'): +class SessionLoadPlan(CoordinationWork, DeclaredFamily, affix='SessionLoadPlan'): @classmethod def at_ingress(cls, metadata: object) -> 'SessionLoadPlan': return EnsuringSessionLoadPlan() if metadata is None else FieldCodec.decode(cls, metadata) @@ -49,7 +49,7 @@ async def resolve(self, lifecycle: "AttachedSessionLifecycle", thread: "Thread") @dataclass(frozen=True, slots=True) class EnsuringSessionLoadPlan(SessionLoadPlan): async def resolve(self, lifecycle: "AttachedSessionLifecycle", thread: "Thread") -> "Thread": - return await coordinator.Coordination.run_worker(partial( + return await self.run_worker(partial( lifecycle.comms.owners.ensure_owner, thread.name, agent_bin=lifecycle.effects.inputs.configuration.launch.binary, agent_args=GivenArguments(lifecycle.effects.inputs.configuration.agent_args.argv), @@ -78,7 +78,7 @@ class ExistingSessionLoadPlan(WitnessedSessionLoadPlan): binding: thread_presentation.LiveThreadOwnerBinding async def resolve(self, lifecycle: "AttachedSessionLifecycle", thread: "Thread") -> "Thread": - entry = (await coordinator.Coordination.run_worker(partial( + entry = (await self.run_worker(partial( lifecycle.comms.registry.entry, thread.name, ))).entry if thread.incarnation != self.binding.owner.incarnation: From a325863122dffc0f68dddaf399343c3aa3187a78 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Mon, 2 Nov 2026 07:27:43 -0500 Subject: [PATCH 09/10] Retain nominal notification keys in observation declarations --- src/agent_comms/ui_model/observation.py | 11 ++++++----- 1 file changed, 6 insertions(+), 5 deletions(-) diff --git a/src/agent_comms/ui_model/observation.py b/src/agent_comms/ui_model/observation.py index 8cab7f71e7..2046316991 100644 --- a/src/agent_comms/ui_model/observation.py +++ b/src/agent_comms/ui_model/observation.py @@ -59,6 +59,7 @@ from agent_comms.message_notification import NotificationWindow from agent_comms.notification_read import NotificationsRead, UnreadableNotificationsRead from agent_comms.message_reference import MessageReference +from agent_comms.nominal.field_codec import MessageId from agent_comms.nominal.fold import Absorbs, Accumulates, Fold from agent_comms.notification_reads import NotificationRead from agent_comms.read_basis import DisplayCut, NoDisplaySelection @@ -593,7 +594,7 @@ def read(self, comms: Comms) -> NotificationWindow: return self.operation.apply(comms, self.display) @property - def keys(self) -> tuple[tuple[int, str], ...]: + def keys(self) -> tuple[tuple[int, MessageId], ...]: """The outcome keys this read shows, as windows key their outcomes.""" return tuple((reference.seq, reference.message_id) for reference in self.operation.references) @@ -791,7 +792,7 @@ class NotificationBatch(BatchRead): observed: NotificationAnswer @cached_property - def keys(self) -> frozenset[tuple[int, str]]: + def keys(self) -> frozenset[tuple[int, MessageId]]: return frozenset(self.request.keys) def settled(self, family: ReadNotifications, current: list[NotificationBatch], @@ -812,9 +813,9 @@ def __init__(self, request: ObserveNotifications): self.batches: tuple[NotificationBatch, ...] = () self.answered: dict[ReadNotifications, tuple[NotificationBatch, ...]] = {} - def wanted(self) -> dict[ReadNotifications, frozenset[tuple[int, str]]]: + def wanted(self) -> dict[ReadNotifications, frozenset[tuple[int, MessageId]]]: """Every key some open request shows, by the request family that reads it.""" - wanted: dict[ReadNotifications, frozenset[tuple[int, str]]] = {} + wanted: dict[ReadNotifications, frozenset[tuple[int, MessageId]]] = {} for request in self.request.reads: wanted[request.family] = wanted.get(request.family, frozenset()) | frozenset(request.keys) return wanted @@ -1199,7 +1200,7 @@ class NotificationAnswer(Observed, Fold): @dataclass(frozen=True) class NotificationsObserved(Accumulates, NotificationAnswer): - outcomes: Mapping[tuple[int, str], tuple[MessageNotification, ...]] + outcomes: Mapping[tuple[int, MessageId], tuple[MessageNotification, ...]] def including(self, batch: NotificationsObserved) -> NotificationsObserved: """This request's answer with a readable batch's outcomes for the keys it shows.""" From 6464ef0951b7aff41d3526c57254b6ae4533206c Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Mon, 2 Nov 2026 08:40:09 -0500 Subject: [PATCH 10/10] Capture the resolved combined sources Agent-Comms-Integration: {"carrier":{"head":"f45f37216e964af3bba2d7db7609bfe7d8f8077c","pr":3857,"repo":"openhcsdev/agent-comms"},"kind":"combined","sources":[{"baseline":{"commit":"ed110cc79cb4bf44964b85e1f41c241ae71880e3","repository":"openhcsdev/agent-comms"},"candidate_tree":"8b0f9a1c3ef8743d2dbb50a9cb8a95e85442b207","heads":[{"head":"a325863122dffc0f68dddaf399343c3aa3187a78","pr":3848,"repo":"openhcsdev/agent-comms"}],"kind":"local_integration"},{"baseline":{"commit":"b67c97e5c3ad1eda5c06a464fdc578038d01e3e8","repository":"openhcsdev/formbridge"},"candidate_tree":"8632c131ec6c529cae78de44d86e191dbbbd844a","heads":[],"kind":"merged_integration"},{"baseline":{"commit":"60138d533ef0758c5f093d9110ba9f5656f0934a","repository":"openhcsdev/metaclass-registry"},"candidate_tree":"3c11508a7bf34cfbe79964127305e056227b6b56","heads":[],"kind":"merged_integration"},{"baseline":{"commit":"da87b77f22bd032bf7cd6d52630f684fa4a2b320","repository":"openhcsdev/textual"},"candidate_tree":"69923b8856c78a1053ca02d045f5b23e875a2c5b","heads":[],"kind":"merged_integration"},{"baseline":{"commit":"51cf2655dd50b5b32bb707fcfacda56381c2eae3","repository":"openhcsdev/toad"},"candidate_tree":"9ea61fcfb38bbc9e0e9e1076f04844a6af6bbfb6","carrier":{"head":"ebaad6cf6633141c3a06b8671ee0287bd6a3330f","pr":1407,"repo":"openhcsdev/toad"},"heads":[{"head":"e7e409268d3469f697f0a69149933f529dca11b0","pr":1389,"repo":"openhcsdev/toad"}],"kind":"resolved_integration"},{"baseline":{"commit":"ba614bcb90b38c1f70789059a3c6aef8c017971b","repository":"openhcsdev/ui-theme"},"candidate_tree":"6a6cbe340cdd660e3d59cd086b15566d8836a0a2","heads":[],"kind":"merged_integration"}]}