From 0f4a395e4eeea9c6226129fa84adc54890ae710c Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Fri, 23 Oct 2026 04:47:23 -0400 Subject: [PATCH 1/3] Preserve retained goal waits across edits during reply review --- src/agent_comms/goal_management.py | 11 ++-- src/agent_comms/goal_waits.py | 13 ++++- tests/test_goal_reply_lock_scope.py | 85 ++++++++++++++++++++++++++++- 3 files changed, 102 insertions(+), 7 deletions(-) diff --git a/src/agent_comms/goal_management.py b/src/agent_comms/goal_management.py index 2e62c307a0..b288290142 100644 --- a/src/agent_comms/goal_management.py +++ b/src/agent_comms/goal_management.py @@ -395,18 +395,19 @@ def consume_reply_wait(self, owner: RegistryOwner, reference: MessageReference) with self.registry.store.locked(mode=SharedLock()): owner.require_registry(self.registry) review = GoalWaitReview.capture(self, owner.thread.incarnation.require_recorded()) - current = review.owner - goal = current.continuation_goal(owner.thread) - if not review.wait.owner_matches(current.incarnation.require_recorded()): + incarnation = review.owner.incarnation.require_recorded() + if not review.wait.owner_matches(incarnation): return False original, = self.bus.bus.deliveries_for_references((reference,)) - if not review.wait.matches(original, current.incarnation.require_recorded()): + if not review.wait.matches(original, incarnation): return False with self.registry.store.locked(): owner.require_registry(self.registry) if not review.current_for(self): return False - return any(review.wait.current_for(actual, current.incarnation.require_recorded()) + current, = self.registry.snapshot().threads_for(incarnation) + goal = current.continuation_goal(owner.thread) + return any(review.wait.current_for(actual, incarnation) and review.wait.clear(self.waits) for actual in goal) def direct(self, name: str, message: Message) -> None: diff --git a/src/agent_comms/goal_waits.py b/src/agent_comms/goal_waits.py index 7ff14d348c..fcc33dbd1e 100644 --- a/src/agent_comms/goal_waits.py +++ b/src/agent_comms/goal_waits.py @@ -273,13 +273,20 @@ def current_for(self, goals: Goals) -> bool: """Caller holds the publication lock; no reply decoding belongs in this check.""" snapshot = goals.registry.snapshot() return any( - owner.goal == self.owner.goal + self._current_goal(owner) and owner.turn_lease == self.turn and goals.waits.resolve(owner.goal) == self.wait for owner in snapshot.threads_for(self.owner.incarnation.require_recorded()) if self._same_thread(snapshot, owner, self.owner) ) + def _current_goal(self, owner: Thread) -> bool: + """A retained wait follows edits of its active goal, not a frozen goal value.""" + return any( + self.wait.current_for(goal, owner.incarnation.require_recorded()) + for goal in owner.goal.active() + ) + def _same_thread(self, snapshot: RegistrySnapshot, actual: Thread, expected: Thread) -> bool: """Ignore display metadata, but retain the original process and input/turn epochs.""" return ( @@ -334,6 +341,10 @@ class GoalInputReview(GoalWaitReview): unknown: tuple[StoredInput, ...] eligible_keys: frozenset[str] + def _current_goal(self, owner: Thread) -> bool: + """Publishing an input review still requires the exact prepared goal.""" + return owner.goal == self.owner.goal + def current_for(self, goals: Goals) -> bool: if not super(GoalInputReview, self).current_for(goals): return False diff --git a/tests/test_goal_reply_lock_scope.py b/tests/test_goal_reply_lock_scope.py index daf0e22269..6810162cb7 100644 --- a/tests/test_goal_reply_lock_scope.py +++ b/tests/test_goal_reply_lock_scope.py @@ -21,11 +21,12 @@ SetGoalAction, StandbyGoalAction, ) -from agent_comms.goal_waits import GoalWaitReview, RecordedGoalWait +from agent_comms.goal_waits import GoalWaitReview, NoGoalWait, RecordedGoalWait from agent_comms.input_disposition import InputDispositions from agent_comms.messages import ConversationRole, NoticeRole from agent_comms.store_files import store_lock_path from agent_comms.thread_text import GivenThreadText +from agent_comms.thread_status import StoppedThreadStatus from agent_comms.tool_caller import ThreadCaller from agent_comms.tools import CommsGoalTool from goal_owner_fixture import canonical_goal_wire @@ -332,3 +333,85 @@ def test_display_edit_during_terminal_review_does_not_strand_wait(reviewing, ren ) assert reviewing.registry.require("owner").title == title assert reviewing.goals.goal_wait("owner") != captured.wait + + +def edit_waiting_goal(comms): + comms.goals.update_goal("owner", EditGoalAction(text="Keep this newer objective")) + + +def edit_waiting_progress(comms): + owner = comms.registry.require("owner") + goal = owner.goal.require() + comms.registry.register(replace( + owner, goal=replace(goal, progress="Keep this newer progress", revision=goal.revision + 1), + )) + + +def edit_waiting_title(comms): + owner = comms.registry.require("owner") + comms.registry.register(replace(owner, title=GivenThreadText("Keep this newer title"))) + + +def stop_unrelated_thread(comms): + comms.registry.register(comms.registry.require("other"), StoppedThreadStatus()) + + +def release_terminal_wait(comms): + fence = comms.agents.finish_turn(comms.registry.require("child").require_turn_lease()) + return partial(comms.goals.release_waits_after_terminal_turn, fence) + + +def recover_closed_wait(comms): + comms.agents.finish_turn(comms.registry.require("child").require_turn_lease()) + comms.registry.unregister("child") + return partial(comms.goals.recover_closed_goal_wait, "owner") + + +@pytest.mark.parametrize("release", (release_terminal_wait, recover_closed_wait)) +@pytest.mark.parametrize("change", ( + edit_waiting_goal, edit_waiting_progress, edit_waiting_title, stop_unrelated_thread, +)) +def test_retained_wait_releases_in_same_call_after_concurrent_edit(reviewing, release, change): + reviewing.registry.declare(reviewing.registry.new_thread("other", frozenset(), str(reviewing.root))) + reviewing.goals.update_goal("owner", standby(reviewing)) + wait = reviewing.goals.goal_wait("owner") + current_goal = reviewing.registry.require("owner").goal.require() + release_once = release(reviewing) + + def change_after_snapshot(): + nonlocal current_goal + change(reviewing) + assert reviewing.goals.goal_wait("owner") == wait + current_goal = reviewing.registry.require("owner").goal.require() + + reviewing.bus.bus = AfterReadBus(reviewing.bus.bus.path, change_after_snapshot) + assert release_once() == ("owner",) + assert reviewing.bus.bus.changed + assert reviewing.goals.goal_wait("owner") == NoGoalWait() + published = reviewing.registry.require("owner").goal.require() + assert published.revision == current_goal.revision + 1 + assert published.progress.startswith(current_goal.progress) + assert "Standby was released" in published.progress + assert replace(published, progress=current_goal.progress, revision=current_goal.revision) == current_goal + + +@pytest.mark.parametrize("change", (edit_waiting_goal, edit_waiting_progress)) +def test_completed_reply_consumes_retained_wait_after_concurrent_goal_edit(reviewing, change): + witness, reply, wait = completed_reply(reviewing) + reviewing.bus.bus = AfterReadBus(reviewing.bus.bus.path, partial(change, reviewing)) + assert reviewing.goals.consume_reply_wait(witness, reply.reference) + assert reviewing.bus.bus.changed + assert reviewing.goals.goal_wait("owner") == NoGoalWait() + assert reviewing.registry.require("owner").goal.require().revision == wait.revision + 1 + + +def test_new_dependency_turn_during_silence_read_keeps_wait(reviewing): + reviewing.goals.update_goal("owner", standby(reviewing)) + wait = reviewing.goals.goal_wait("owner") + fence = reviewing.agents.finish_turn(reviewing.registry.require("child").require_turn_lease()) + reviewing.bus.bus = AfterReadBus( + reviewing.bus.bus.path, partial(reviewing.agents.begin_turn, "child", "new-child-turn"), + ) + assert reviewing.goals.release_waits_after_terminal_turn(fence) == () + assert reviewing.bus.bus.changed + assert reviewing.goals.goal_wait("owner") == wait From 3efbdf071c2a3f05650a134406e4cb9b33e112ad Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Fri, 23 Oct 2026 04:49:16 -0400 Subject: [PATCH 2/3] Reobserve the same wait after its owner fence changes --- src/agent_comms/goal_waits.py | 16 +++++++++++++--- tests/test_goal_reply_lock_scope.py | 17 +++++++++++++++++ 2 files changed, 30 insertions(+), 3 deletions(-) diff --git a/src/agent_comms/goal_waits.py b/src/agent_comms/goal_waits.py index fcc33dbd1e..60872b4682 100644 --- a/src/agent_comms/goal_waits.py +++ b/src/agent_comms/goal_waits.py @@ -305,6 +305,7 @@ def require_current(self, goals: Goals, owner: Thread) -> None: def publishing_unanswered(self, goals: Goals) -> Iterator[bool]: """Read without EX; hold it only after this owner's received cut still agrees.""" owner = self.owner.incarnation.require_recorded() + review = self while True: with goals.bus.bus.reading() as source: replied = self.wait.has_reply(source, owner) @@ -313,9 +314,18 @@ def publishing_unanswered(self, goals: Goals) -> Iterator[bool]: yield False return with goals.registry.store.locked(): - if not self.current_for(goals): - yield False - return + if not review.current_for(goals): + try: + review = GoalWaitReview.capture(goals, owner) + except UnregisteredThreadError: + yield False + return + if review.wait != self.wait or not review._current_goal(review.owner): + yield False + return + # A new owner/turn fence needs a fresh silence read for this + # same wait. Re-evaluate here; a terminal callback runs once. + continue # The existing indexed sequence query reads at most one metadata row. # Any new candidate needs another unlocked policy read, not a stale release. if goals.bus.bus.received_after(owner.lookup, through_seq): diff --git a/tests/test_goal_reply_lock_scope.py b/tests/test_goal_reply_lock_scope.py index 6810162cb7..674bca7c86 100644 --- a/tests/test_goal_reply_lock_scope.py +++ b/tests/test_goal_reply_lock_scope.py @@ -415,3 +415,20 @@ def test_new_dependency_turn_during_silence_read_keeps_wait(reviewing): assert reviewing.goals.release_waits_after_terminal_turn(fence) == () assert reviewing.bus.bus.changed assert reviewing.goals.goal_wait("owner") == wait + + +def test_terminal_callback_reobserves_same_wait_after_waiter_turn_changes(reviewing): + reviewing.goals.update_goal("owner", standby(reviewing)) + wait = reviewing.goals.goal_wait("owner") + owner = reviewing.registry.require("owner") + fence = reviewing.agents.finish_turn(reviewing.registry.require("child").require_turn_lease()) + + def finish_waiter_turn(): + witness = reviewing.agents.begin_turn("owner", "intervening-owner-turn") + reviewing.agents.finish_turn(witness.thread.require_turn_lease()) + assert reviewing.goals.goal_wait("owner") == wait + + reviewing.bus.bus = AfterReadBus(reviewing.bus.bus.path, finish_waiter_turn) + assert reviewing.goals.release_waits_after_terminal_turn(fence) == ("owner",) + assert reviewing.registry.require("owner").turn_generation > owner.turn_generation + assert reviewing.goals.goal_wait("owner") == NoGoalWait() From e4ddafa329c4bddbba96965ac20e3900c3bb8a38 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Fri, 23 Oct 2026 04:53:28 -0400 Subject: [PATCH 3/3] Report vanished captured turn records as stale owner fences --- src/agent_comms/native_input_owner.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/agent_comms/native_input_owner.py b/src/agent_comms/native_input_owner.py index 4e1d192a83..995e7ceef7 100644 --- a/src/agent_comms/native_input_owner.py +++ b/src/agent_comms/native_input_owner.py @@ -116,7 +116,7 @@ def _require_current( process=ProcessIdentity.capture(os.getpid()), goal_permission=self.goal_permission, ).require_valid() - except ReservationViolationError as error: + except (ReservationViolationError, RelationViolationError) as error: raise StaleFence(f"{reason}: {error}") from error def require_snapshot(self, snapshot: RegistrySnapshot, reason: str) -> None: