From a07aca61b657f4950dd4d921c269509bdf102b2a Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Sun, 1 Nov 2026 19:17:04 -0500 Subject: [PATCH 1/5] 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 2089a32a6e..ce4adb4a04 100644 --- a/src/agent_comms/tools.py +++ b/src/agent_comms/tools.py @@ -460,7 +460,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 20ca488a3b..fde9f840e1 100644 --- a/tests/test_checkout_submodules.py +++ b/tests/test_checkout_submodules.py @@ -478,7 +478,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) @@ -506,8 +506,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) @@ -539,8 +539,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 700a6dd619..498fceb09c 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 @@ -39,13 +40,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: @@ -54,7 +55,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 @@ -92,13 +93,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 fce3df22b2..7c26ab85c0 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, @@ -58,7 +58,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") @@ -85,7 +85,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 d7b8cf3a16..35c022ed0f 100644 --- a/tests/test_registration_inheritance.py +++ b/tests/test_registration_inheritance.py @@ -86,9 +86,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),) @@ -110,7 +110,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 13d03c3ba47704b660be970787b81322319b710e Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Sun, 1 Nov 2026 22:42:55 -0500 Subject: [PATCH 2/5] Keep ACP session and thread identities distinct in project fixtures --- 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 498fceb09c..db68a4733f 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 @@ -30,8 +29,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: @@ -48,16 +47,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() @@ -71,7 +70,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 7c26ab85c0..3b00878b85 100644 --- a/tests/test_project_resume.py +++ b/tests/test_project_resume.py @@ -80,7 +80,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 b3eb0b93b862eed9fe8d1523e507cbadea41cfd2 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Sun, 1 Nov 2026 22:46:12 -0500 Subject: [PATCH 3/5] 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 163d446113f149f0fe9c01ffc5e2cbf777990334 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Sun, 1 Nov 2026 23:18:20 -0500 Subject: [PATCH 4/5] 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 | 24 +++++++++++++++++++----- tests/test_runtime.py | 3 ++- tests/test_transcript_input_display.py | 13 ++++++++++--- 7 files changed, 54 insertions(+), 21 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 5735470e54..abdefac047 100644 --- a/src/agent_comms/session_lifecycle.py +++ b/src/agent_comms/session_lifecycle.py @@ -245,7 +245,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), @@ -351,7 +353,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 602604381e..138e05ae87 100644 --- a/tests/test_acp.py +++ b/tests/test_acp.py @@ -167,8 +167,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) @@ -194,17 +195,30 @@ async def test_snapshot_capability_replays_one_bounded_update( if renamed: agent._comms.threads.rename_managed_thread("proj", "renamed", owner=ProcessIdentity.capture(os.getpid())) sent = [] + destinations = [] + destination = AcpSessionId("snapshot-destination") + thread = agent._comms.registry.require("proj") class Client(SessionUpdateSink): - async def publish_update(self, **kwargs): - sent.append(kwargs["update"]) + async def publish_update(self, *, session_id, update): + destinations.append(session_id) + sent.append(update) return True 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 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 4b9e180e3019fd23c879a6c469781f1fe23c87eb Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Mon, 2 Nov 2026 00:03:52 -0500 Subject: [PATCH 5/5] 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]), )