Release retained goal waits across concurrent goal edits

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:

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 (
@@ -298,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)
@@ -306,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):
@@ -334,6 +351,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

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:

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,102 @@ 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
+
+
+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()