From 530ffc390489b66497bac472c99a5667bc99a36c Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Fri, 23 Oct 2026 02:00:30 -0400 Subject: [PATCH 1/5] Read goal replies outside the registry publication lock --- src/agent_comms/goal_actions.py | 27 +++- src/agent_comms/goal_management.py | 230 ++++++++++++++-------------- src/agent_comms/goal_waits.py | 72 +++++++-- tests/test_goal_reply_lock_scope.py | 186 ++++++++++++++++++++++ 4 files changed, 389 insertions(+), 126 deletions(-) create mode 100644 tests/test_goal_reply_lock_scope.py diff --git a/src/agent_comms/goal_actions.py b/src/agent_comms/goal_actions.py index 6ae4a7902e..9d1c643edf 100644 --- a/src/agent_comms/goal_actions.py +++ b/src/agent_comms/goal_actions.py @@ -41,7 +41,7 @@ PausedGoal, RuntimePause, ) -from .goal_waits import GoalWaits, RecordedGoalWait +from .goal_waits import GoalInputReview, GoalWaits, RecordedGoalWait from .lifecycle_capability import Settled from .owner_requests import OwnerRequestId from .progress_claims import ProgressClaim @@ -267,10 +267,31 @@ class StandbyGoalAction(TransitionGoalAction, ModelInvocable, RuntimeInvocable): def next_state(self, goal: Goal, ctx: GoalActionContext) -> GoalState: return ActiveGoal() + def prepare(self, goals: Goals, name: str, actor: type) -> GoalAction[Goal]: + ctx = GoalActionContext(goals, goals.registry.require(name), actor) + self.expect.check(ctx) + review = goals._goal_input_review(name, ctx.thread.goal.require().id, self.wait_for) + self.expect.check(GoalActionContext(goals, review.owner, actor)) + return PreparedStandbyGoalAction( + expect=replace(self.expect, expected_goal=review.owner.goal), + progress=self.progress, + wait_for=self.wait_for, + reviewed_inputs=self.reviewed_inputs, + review=review, + ) + + +@dataclass(frozen=True, kw_only=True) +class PreparedStandbyGoalAction(StandbyGoalAction, declared_name=None): + """Only the action prepared from real replies can publish this exact wait.""" + + review: GoalInputReview + def before_goal_publish(self, goal: Goal, ctx: GoalActionContext) -> None: + review = self.review + review.require_current(ctx.goals) thread = ctx.thread report_turn = ctx.report_turn - review = ctx.goals._goal_input_review(thread, goal.id, self.wait_for) wait_targets = review.targets aliases = review.owners dispositions = InputDispositions(ctx.goals.root) @@ -340,7 +361,7 @@ def before_goal_publish(self, goal: Goal, ctx: GoalActionContext) -> None: goal.id, uuid4().hex, goal.revision, - ctx.goals.bus.bus.revision(), + review.through_seq, wait_targets, owner=thread.incarnation.require_recorded(), report_turn_id=reporting_lease.stored_turn_id(), diff --git a/src/agent_comms/goal_management.py b/src/agent_comms/goal_management.py index 3f503c998f..4ef1f33178 100644 --- a/src/agent_comms/goal_management.py +++ b/src/agent_comms/goal_management.py @@ -14,7 +14,7 @@ from .agent_activity import AgentActivity from .coordination_errors import StaleFence -from .errors import RelationViolationError +from .errors import RelationViolationError, UnregisteredThreadError from .goal_actions import ( GoalAction, GoalActionContext, @@ -26,6 +26,7 @@ from .goal_history import GoalHistoryEntry from .goal_waits import ( GoalInputReview, + GoalWaitReview, GoalWait, GoalWaits, RecordedGoalWait, @@ -34,6 +35,7 @@ from .native_input_owner import GoalLaunchOwner from .owner_requests import NewestOpenRequest, RequestDirective from .registration import Registration +from .store_files import SharedLock if TYPE_CHECKING: from .agent_events import GoalChange @@ -166,28 +168,30 @@ def dismiss_historical_inputs( def goal_input_review(self, name: str, goal_id: str, wait_for: Sequence[str]) -> StandbyReview: """Project exact review eligibility for one current goal and dependency set.""" - review = self._goal_input_review(self.registry.require(name), goal_id, wait_for) + review = self._goal_input_review(name, goal_id, wait_for) return review.standby_review() def _goal_input_review( - self, thread: Thread, goal_id: str, wait_for: Sequence[str] + self, name: str, goal_id: str, wait_for: Sequence[str] ) -> GoalInputReview: + captured = GoalWaitReview.capture(self, name) + thread = captured.owner + snapshot = captured.registry.snapshot goal = thread.goal.require().require_id(goal_id).require_active() if not wait_for: raise ValueError("Standby requires explicit wait_for thread names.") - resolved = tuple(self.registry.require(target.removeprefix("@")) for target in wait_for) + resolved = tuple(snapshot.require(target.removeprefix("@")) for target in wait_for) if any(target.incarnation == thread.incarnation for target in resolved): raise ValueError("A goal cannot wait for its own thread.") targets = tuple( dict.fromkeys(GoalWaitTarget(target.incarnation.require_recorded()) for target in resolved) ) - snapshot = self.registry.snapshot() - owners = snapshot.names_for(thread.name) + owners = captured.owners dispositions = InputDispositions(self.root) unknown = {row.key: row for row in dispositions.unresolved(owners).unknown(owners)} eligible = set() lookup = thread.incarnation.lookup - after_seq = self.goal_wait(thread.name).review_after_seq( + after_seq = captured.wait.review_after_seq( goal, thread.incarnation.require_recorded(), targets ) sequences = tuple({ @@ -197,6 +201,7 @@ def _goal_input_review( senders = tuple(target.incarnation.lookup for target in targets) replies: dict[str, Message] = {} with self.bus.bus.reading() as source: + through_seq = source.revision() # Preserve older UNKNOWN rows; new inputs begin at this wait's own cursor. # Both selections use the same frozen recipient and sender projections. page = source.received( @@ -234,9 +239,12 @@ def _goal_input_review( unknown[key] = pending eligible.add(key) return GoalInputReview( - goal_id, + captured.registry, + thread, + captured.wait, + captured.turn, + through_seq, targets, - owners, tuple(sorted(unknown.values(), key=lambda row: row.order)), frozenset(eligible), ) @@ -252,106 +260,102 @@ def goal_wait(self, name: str) -> GoalWait: return self.waits.resolve(goal) def recover_closed_goal_wait(self, name: str) -> tuple[str, ...]: - """Release one stranded standby without replaying a dependency input. - - This is an owner-side liveness check, not an agent turn. A subsequent - scheduler pass may continue the still-active goal only if its private - attempt ledger has a fresh READY grant. - """ - with self.registry.store.locked(): - snapshot = self.registry.snapshot() - canonical = snapshot.canonical_name(name) - owner = snapshot.threads.get(canonical) - if owner is None or self.agents.is_working(owner.incarnation, snapshot): - return () - return tuple(member for goal in owner.goal.active() - for member in self._recover_closed_wait(goal, owner, snapshot, canonical)) - - def _recover_closed_wait(self, goal: Goal, owner: Thread, snapshot: RegistrySnapshot, - canonical: str) -> tuple[str, ...]: - waits = self.waits - rows = waits.read() - wait = rows.get(goal.id) - if wait is None or not wait.current_for(goal, owner.incarnation.require_recorded()): - return () - closed = GoalWaits.closed_wait_group(canonical, wait.targets, rows, snapshot, self.agents) - if not closed: + """Read silence outside the registry, then release only the observed wait.""" + try: + review = GoalWaitReview.capture(self, name) + except UnregisteredThreadError: return () - # An unreadable bus remains a drain diagnostic, never evidence of silence. - if wait.has_reply(self.bus.bus, owner.incarnation.require_recorded()): + owner = review.owner + snapshot = review.registry.snapshot + if self.agents.is_working(owner.incarnation, snapshot) or not any( + review.wait.current_for(goal, owner.incarnation.require_recorded()) + for goal in owner.goal.active() + ): return () - names = ", ".join(f"@{member}" for member in closed) - note = ( - f"Dependency wait group ({names}) has no independent worker. " - "Standby was released; inspect dependencies and continue useful work." - ) - progress = f"{goal.progress}\n\n{note}" if goal.progress else note - self.registry.register( - replace(owner, goal=replace(goal, progress=progress, revision=goal.revision + 1)), - snapshot.status(canonical), + closed = GoalWaits.closed_wait_group( + owner.name, review.wait.targets, self.waits.read(), snapshot, self.agents, ) - if not waits.clear(goal.id, wait_id=wait.wait_id): + if not closed: return () - return closed + # A failed read is a diagnostic, never evidence of silence. + for through_seq in review.unanswered(self.bus.bus): + with self.registry.store.locked(): + if not review.current_for(self): + return () + snapshot = self.registry.snapshot() + if self.agents.is_working(owner.incarnation, snapshot): + return () + closed = GoalWaits.closed_wait_group( + owner.name, review.wait.targets, self.waits.read(), snapshot, self.agents, + ) + if not closed: + return () + names = ", ".join(f"@{member}" for member in closed) + note = ( + f"Dependency wait group ({names}) has no independent worker. " + "Standby was released; inspect dependencies and continue useful work." + ) + if self._release_reviewed_wait(review, through_seq, note): + return closed + return () def release_waits_after_terminal_turn(self, fence: TurnCompletion) -> tuple[str, ...]: - """Release waits after the latest exact, still-idle child turn. - - ACP invokes this after terminal publication, never at the earlier UI - `settled` event. A later turn (even already finished) invalidates the - old fence; a rename or unrelated metadata edit does not. - """ - with self.registry.store.locked(): - snapshot = self.registry.snapshot() - sources = fence.idle_sources(snapshot) - if not sources: - return () - (source,) = sources - canonical = source.name - released: list[str] = [] - for wait in self.waits.targeting(source.incarnation.require_recorded()): - for owner in snapshot.threads_for(wait.owner): - if any( - self._release_finished_wait( - goal, - owner, - snapshot, - canonical, - wait.goal_id, - wait, - ) - for goal in owner.goal.active() - ): - released.append(owner.name) - return tuple(released) - - def _release_finished_wait( - self, - goal: Goal, - owner: Thread, - snapshot: RegistrySnapshot, - canonical: str, - goal_id: str, - wait: RecordedGoalWait, - ) -> bool: - if goal.id != goal_id or not wait.current_for(goal, owner.incarnation.require_recorded()): + """A completed turn may release silence observed outside the registry lock.""" + snapshot = self.registry.snapshot() + sources = fence.idle_sources(snapshot) + if not sources: + return () + source, = sources + released: list[str] = [] + for wait in self.waits.targeting(source.incarnation.require_recorded()): + for owner in snapshot.threads_for(wait.owner): + try: + review = GoalWaitReview.capture(self, owner.name) + except UnregisteredThreadError: + continue + if review.wait == wait and self._release_finished_wait(review, fence): + released.append(review.owner.name) + return tuple(released) + + def _release_finished_wait(self, review: GoalWaitReview, fence: TurnCompletion) -> bool: + wait, owner = review.wait, review.owner + if not any(wait.current_for(goal, owner.incarnation.require_recorded()) + for goal in owner.goal.active()): return False - if any(self.agents.is_working(target.incarnation, snapshot) for target in wait.targets): + if any(self.agents.is_working(target.incarnation, review.registry.snapshot) + for target in wait.targets): return False - if wait.has_reply(self.bus.bus, owner.incarnation.require_recorded()): + for through_seq in review.unanswered(self.bus.bus): + with self.registry.store.locked(): + if not review.current_for(self): + return False + current = self.registry.snapshot() + sources = fence.idle_sources(current) + if not sources or any(self.agents.is_working(target.incarnation, current) + for target in wait.targets): + return False + actual_source, = sources + note = ( + f"Declared dependency @{actual_source.name} finished without a qualifying direct " + "reply, and no declared dependency has an active turn. " + "Standby was released; inspect messages and UNKNOWN inputs, then " + "continue independent work or redelegate. No input was replayed." + ) + return self._release_reviewed_wait(review, through_seq, note) + return False + + def _release_reviewed_wait(self, review: GoalWaitReview, through_seq: int, note: str) -> bool: + """Caller holds the registry; a new append invalidates the silence just read.""" + if self.bus.bus.revision() != through_seq: return False - diagnostic = ( - f"Declared dependency @{canonical} finished without a qualifying direct " - "reply, and no declared dependency has an active turn. " - "Standby was released; inspect messages and UNKNOWN inputs, then " - "continue independent work or redelegate. No input was replayed." + owner = review.owner + goal = owner.goal.require_active() + progress = f"{goal.progress}\n\n{note}" if goal.progress else note + self.registry.register( + replace(owner, goal=replace(goal, progress=progress, revision=goal.revision + 1)), + review.registry.snapshot.status(owner.name), ) - progress = f"{goal.progress}\n\n{diagnostic}" if goal.progress else diagnostic - # Persist the explanation before releasing the exact wait. - continued_goal = replace(goal, progress=progress, revision=goal.revision + 1) - self.registry.register(replace(owner, goal=continued_goal), snapshot.status(owner.name)) - self.waits.clear(goal.id, wait_id=wait.wait_id) - return True + return review.wait.clear(self.waits) def goal_execution(self, name: str) -> GoalExecution: return self._goal_snapshot(name).execution @@ -395,25 +399,23 @@ def consume_goal_wait(self, name: str, wait_id: str) -> bool: return goal.clear_wait(self.waits, wait_id=wait_id) def consume_reply_wait(self, owner: RegistryOwner, reference: MessageReference) -> bool: - """A completed native input consumes only its original dependency wait. - - Selected handling calls this after corroborated completion, never after - fetching, triage preparation or a failed/UNKNOWN native send. - """ - with self.registry.store.locked(): + """Read the exact completed delivery before checking its original owner fence.""" + with self.registry.store.locked(mode=SharedLock()): owner.require_registry(self.registry) - current = self.registry.require(owner.thread.name) - goal = current.continuation_goal(owner.thread) - return any(self._consume_reply_wait(actual, current, reference) for actual in goal) - - def _consume_reply_wait(self, goal: Goal, current: Thread, reference: MessageReference) -> bool: - wait = self.waits.for_goal(goal, self.waits.read()) - if not wait.owner_matches(current.incarnation.require_recorded()): + review = GoalWaitReview.capture(self, owner.thread.name) + current = review.owner + goal = current.continuation_goal(owner.thread) + if not review.wait.owner_matches(current.incarnation.require_recorded()): return False original, = self.bus.bus.deliveries_for_references((reference,)) - if not wait.matches(original, current.incarnation.require_recorded()): + if not review.wait.matches(original, current.incarnation.require_recorded()): return False - return wait.consume_reply(self.waits, goal.id) + 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()) + and review.wait.clear(self.waits) for actual in goal) def direct(self, name: str, message: Message) -> None: """A plain owner message directs the recipient's newest open request.""" diff --git a/src/agent_comms/goal_waits.py b/src/agent_comms/goal_waits.py index e838547762..98e82c33ad 100644 --- a/src/agent_comms/goal_waits.py +++ b/src/agent_comms/goal_waits.py @@ -8,7 +8,7 @@ from pathlib import Path from typing import TYPE_CHECKING, Annotated, ClassVar -from .bus import Bus +from .bus import BusReader from .bus_publication import Delivered from .bus_tables import DirectReplySource from .declared_family import DeclaredFamily @@ -17,14 +17,19 @@ from .goals import Goal, GoalDeclaration from .input_attempt import StoredInput, UnresolvedInput from .locked_store import WireWakeFile +from .private_path import FileRevision +from .store_files import SharedLock from .thread_identity import RecordedThreadIncarnation from .thread_owned_state import ThreadOwnedState +from .turn_lease import TurnLease from .typed_table import Column, DurableStore, ForeignKey, Index, TypedTable if TYPE_CHECKING: from .agent_activity import AgentActivity from .registry_document import RegistrySnapshot from .threads import Thread + from .goal_management import Goals + from .registry_store import RegistryRevision class GoalWait(ABC): @@ -62,7 +67,13 @@ def review_after_seq( def consume_dependency(self, comms, canonical: str) -> bool: return True - def consume_reply(self, waits: GoalWaits, goal_id: str) -> bool: + def clear(self, waits: GoalWaits) -> bool: + return False + + def current_for(self, goal: Goal, owner: RecordedThreadIncarnation) -> bool: + return False + + def has_reply(self, bus: BusReader, owner: RecordedThreadIncarnation) -> bool: return False @abstractmethod @@ -145,8 +156,8 @@ def owner_matches(self, owner: RecordedThreadIncarnation) -> bool: def consume_dependency(self, comms, canonical: str) -> bool: return comms.goals.consume_goal_wait(canonical, self.wait_id) - def consume_reply(self, waits: GoalWaits, goal_id: str) -> bool: - return waits.clear(goal_id, wait_id=self.wait_id) + def clear(self, waits: GoalWaits) -> bool: + return waits.clear(self.goal_id, wait_id=self.wait_id) def execution(self, source: GoalExecutionSource) -> GoalExecution: targets = tuple( @@ -187,7 +198,7 @@ def review_after_seq(self, goal, owner, targets) -> int: return self.after_seq return super().review_after_seq(goal, owner, targets) - def has_reply(self, bus: Bus, owner: RecordedThreadIncarnation) -> bool: + def has_reply(self, bus: BusReader, owner: RecordedThreadIncarnation) -> bool: if not self.owner_matches(owner): return False return any( @@ -243,12 +254,55 @@ class StandbyReview: @dataclass(frozen=True, slots=True) -class GoalInputReview: - """Ephemeral dependency review projection derived from bus and input authority.""" +class GoalWaitReview: + """The existing registry, turn and wait observed before reading reply history.""" - goal_id: str + registry: RegistryRevision + owner: Thread + wait: GoalWait + turn: TurnLease + + @classmethod + def capture(cls, goals: Goals, name: str) -> GoalWaitReview: + with goals.registry.store.locked(mode=SharedLock()): + registry = goals.registry.store.read_revision_unlocked() + owner = registry.snapshot.require(name) + return cls(registry, owner, goals.waits.resolve(owner.goal), owner.turn_lease) + + def current_for(self, goals: Goals) -> bool: + """Caller holds the publication lock; no reply decoding belongs in this check.""" + return ( + FileRevision.capture(goals.registry.store.path) == self.registry.revision + and self.owner.turn_lease == self.turn + and goals.waits.resolve(self.owner.goal) == self.wait + ) + + def require_current(self, goals: Goals) -> None: + if not self.current_for(goals): + raise ValueError("Goal, dependency or wait changed during reply review; inspect it again.") + + def unanswered(self, bus: BusReader) -> tuple[int, ...]: + """The captured bus cut with no qualifying reply, or no such cut.""" + with bus.reading() as source: + if self.wait.has_reply(source, self.owner.incarnation.require_recorded()): + return () + return (source.revision(),) + + @property + def goal_id(self) -> str: + return self.owner.goal.require().id + + @property + def owners(self) -> frozenset[str]: + return self.registry.snapshot.names_for(self.owner.name) + + +@dataclass(frozen=True, slots=True) +class GoalInputReview(GoalWaitReview): + """Reply eligibility and the exact bus cut inspected before publication.""" + + through_seq: int targets: tuple[GoalWaitTarget, ...] - owners: frozenset[str] unknown: tuple[StoredInput, ...] eligible_keys: frozenset[str] diff --git a/tests/test_goal_reply_lock_scope.py b/tests/test_goal_reply_lock_scope.py new file mode 100644 index 0000000000..6cb67ce0c9 --- /dev/null +++ b/tests/test_goal_reply_lock_scope.py @@ -0,0 +1,186 @@ +"""Reply reads leave the registry free, and publication keeps its observed state.""" + +import os +import subprocess +import sys +from contextlib import contextmanager + +import pytest + +from agent_comms.bus import Bus +from agent_comms.bus_publication import Delivered +from agent_comms.bus_tables import ReceivedSource +from agent_comms.child_process import ProcessIdentity +from agent_comms.goal_actions import ( + EditGoalAction, + GoalActionContext, + GoalPrecondition, + RuntimeInvocable, + SetGoalAction, + StandbyGoalAction, +) +from agent_comms.goal_waits import GoalWaitReview, RecordedGoalWait +from agent_comms.input_disposition import InputDispositions +from agent_comms.native_input_owner import RegistryOwner +from agent_comms.store_files import store_lock_path +from goal_owner_fixture import canonical_goal_wire + + +class RegistryProbeBus(Bus): + """A real SQLite read asks another process to take the real registry lock.""" + + def __init__(self, path, registry_path): + super().__init__(path) + self.registry_lock = store_lock_path(registry_path) + self.checks = [] + + def probe(self, statement): + if self.checks or not any(table in statement for table in ( + ReceivedSource.declared_name, Delivered.table.declared_name, + )): + return + checked = subprocess.run( + [sys.executable, "-c", ( + "import fcntl, sys\n" + "with open(sys.argv[1], 'rb') as lock:\n" + " try: fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)\n" + " except BlockingIOError: sys.exit(11)\n" + ), str(self.registry_lock)], + capture_output=True, text=True, timeout=5, + ) + self.checks.append(checked.returncode) + + @contextmanager + def observing(self): + with super().observing() as db: + db.set_trace_callback(self.probe) + try: + yield db + finally: + db.set_trace_callback(None) + + +@pytest.fixture +def reviewing(tmp_path): + comms = canonical_goal_wire(tmp_path / "wire") + for name in ("owner", "child"): + comms.registry.declare(comms.registry.new_thread( + name, frozenset(), str(tmp_path), + process_identity=ProcessIdentity.capture(os.getpid()), + )) + comms.agents.begin_turn("child", "child-work") + comms.goals.update_goal("owner", SetGoalAction(text="Review the child's work")) + return comms + + +@pytest.fixture +def probed(reviewing): + reviewing.bus.bus = RegistryProbeBus(reviewing.bus.bus.path, reviewing.registry.store.path) + return reviewing + + +def standby(comms, reviewed_inputs=()): + return StandbyGoalAction( + expect=GoalPrecondition(goal_id=comms.registry.require("owner").goal.require().id), + wait_for=("child",), reviewed_inputs=reviewed_inputs, + ) + + +def test_standby_reads_replies_without_holding_registry(probed): + probed.goals.update_goal("owner", standby(probed)) + assert probed.bus.bus.checks == [0] + + +def test_terminal_release_reads_replies_without_holding_registry(probed): + probed.goals.update_goal("owner", standby(probed)) + probed.bus.bus.checks.clear() + fence = probed.agents.finish_turn(probed.registry.require("child").require_turn_lease()) + assert probed.goals.release_waits_after_terminal_turn(fence) == ("owner",) + assert probed.bus.bus.checks == [0] + + +def test_closed_wait_recovery_reads_replies_without_holding_registry(probed): + probed.goals.update_goal("owner", standby(probed)) + probed.agents.finish_turn(probed.registry.require("child").require_turn_lease()) + probed.registry.unregister("child") + probed.bus.bus.checks.clear() + assert probed.goals.recover_closed_goal_wait("owner") == ("owner",) + assert probed.bus.bus.checks == [0] + + +def test_completed_reply_is_decoded_without_holding_registry(probed): + probed.goals.update_goal("owner", standby(probed)) + reply = probed.messaging.send("child", "owner", "Finished").message + probed.bus.bus.checks.clear() + owner = probed.registry.require("owner") + witness = RegistryOwner( + thread=owner, claimed_generation=probed.registry.snapshot().claim_generation(owner.name), + ) + assert probed.goals.consume_reply_wait(witness, reply.reference) + assert probed.bus.bus.checks == [0] + + +def test_reply_appended_between_review_and_publication_is_still_eligible(reviewing): + cut = reviewing.bus.bus.revision() + prepared = standby(reviewing).prepare(reviewing.goals, "owner", RuntimeInvocable) + reply = reviewing.messaging.send("child", "owner", "Arrived during review").message + with reviewing.registry.store.locked(): + prepared.apply(GoalActionContext( + reviewing.goals, reviewing.registry.require("owner"), RuntimeInvocable, + )) + wait = reviewing.goals.goal_wait("owner") + assert wait.after_seq == cut < reply.seq + assert wait.has_reply(reviewing.bus.bus, reviewing.registry.require("owner").incarnation) + review = reviewing.goals.goal_input_review("owner", wait.goal_id, ("child",)) + assert review.reviewed_inputs == (f"bus:{reply.seq}",) + + +@pytest.mark.parametrize("change", ("goal", "wait", "turn", "dependency")) +def test_stale_review_never_writes_inputs_or_replaces_newer_wait(reviewing, change): + reply = reviewing.messaging.send("child", "owner", "Review this reply").message + prepared = standby(reviewing, (f"bus:{reply.seq}",)).prepare( + reviewing.goals, "owner", RuntimeInvocable, + ) + owner = reviewing.registry.require("owner") + goal = owner.goal.require() + if change == "goal": + reviewing.goals.update_goal("owner", EditGoalAction(text="A different objective")) + elif change == "wait": + target = reviewing.registry.require("child") + reviewing.goals.waits.record(RecordedGoalWait( + goal.id, "newer-wait", goal.revision, reply.seq, prepared.review.targets, + owner=owner.incarnation.require_recorded(), + target_turn_generations=(target.require_turn_lease().identity.generation,), + report_turn_id=None, report_turn_generation=None, + )) + elif change == "turn": + reviewing.agents.begin_turn("owner", "new-owner-turn") + else: + reviewing.registry.unregister("child") + reviewing.registry.remove("child") + reviewing.registry.declare(reviewing.registry.new_thread( + "child", frozenset(), owner.worktree, + process_identity=ProcessIdentity.capture(os.getpid()), + )) + inputs = InputDispositions(reviewing.root) + before = inputs.path.read_bytes() if inputs.path.exists() else None + wait_before = reviewing.goals.goal_wait("owner") + with reviewing.registry.store.locked(), pytest.raises(ValueError): + prepared.apply(GoalActionContext( + reviewing.goals, reviewing.registry.require("owner"), RuntimeInvocable, + )) + assert (inputs.path.read_bytes() if inputs.path.exists() else None) == before + assert reviewing.goals.goal_wait("owner") == wait_before + + +def test_late_reply_invalidates_captured_silence(reviewing): + reviewing.goals.update_goal("owner", standby(reviewing)) + reviewing.agents.finish_turn(reviewing.registry.require("child").require_turn_lease()) + captured = GoalWaitReview.capture(reviewing.goals, "owner") + cut, = captured.unanswered(reviewing.bus.bus) + reviewing.messaging.send("child", "owner", "A reply after the silence check") + with reviewing.registry.store.locked(): + assert captured.current_for(reviewing.goals) + assert not reviewing.goals._release_reviewed_wait(captured, cut, "stale silence") + assert reviewing.goals.goal_wait("owner") == captured.wait + assert reviewing.registry.require("owner").goal == captured.owner.goal From e7b5a85e9715a51d8c0ce308fd73155f4a79d623 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Fri, 23 Oct 2026 02:55:28 -0400 Subject: [PATCH 2/5] Fence goal reply publication by recipient and captured owner state --- src/agent_comms/goal_actions.py | 2 +- src/agent_comms/goal_management.py | 109 ++++++++++--------- src/agent_comms/goal_waits.py | 84 +++++++++++---- tests/test_goal_reply_lock_scope.py | 156 ++++++++++++++++++++++++---- 4 files changed, 253 insertions(+), 98 deletions(-) diff --git a/src/agent_comms/goal_actions.py b/src/agent_comms/goal_actions.py index 9d1c643edf..a2835e8f18 100644 --- a/src/agent_comms/goal_actions.py +++ b/src/agent_comms/goal_actions.py @@ -289,7 +289,7 @@ class PreparedStandbyGoalAction(StandbyGoalAction, declared_name=None): def before_goal_publish(self, goal: Goal, ctx: GoalActionContext) -> None: review = self.review - review.require_current(ctx.goals) + review.require_current(ctx.goals, ctx.thread) thread = ctx.thread report_turn = ctx.report_turn wait_targets = review.targets diff --git a/src/agent_comms/goal_management.py b/src/agent_comms/goal_management.py index 4ef1f33178..6067f5592f 100644 --- a/src/agent_comms/goal_management.py +++ b/src/agent_comms/goal_management.py @@ -176,7 +176,7 @@ def _goal_input_review( ) -> GoalInputReview: captured = GoalWaitReview.capture(self, name) thread = captured.owner - snapshot = captured.registry.snapshot + snapshot = captured.snapshot goal = thread.goal.require().require_id(goal_id).require_active() if not wait_for: raise ValueError("Standby requires explicit wait_for thread names.") @@ -190,9 +190,10 @@ def _goal_input_review( dispositions = InputDispositions(self.root) unknown = {row.key: row for row in dispositions.unresolved(owners).unknown(owners)} eligible = set() - lookup = thread.incarnation.lookup + incarnation = thread.incarnation.require_recorded() + lookup = incarnation.lookup after_seq = captured.wait.review_after_seq( - goal, thread.incarnation.require_recorded(), targets + goal, incarnation, targets ) sequences = tuple({ sequence for row in unknown.values() @@ -239,7 +240,7 @@ def _goal_input_review( unknown[key] = pending eligible.add(key) return GoalInputReview( - captured.registry, + captured.snapshot, thread, captured.wait, captured.turn, @@ -266,7 +267,7 @@ def recover_closed_goal_wait(self, name: str) -> tuple[str, ...]: except UnregisteredThreadError: return () owner = review.owner - snapshot = review.registry.snapshot + snapshot = review.snapshot if self.agents.is_working(owner.incarnation, snapshot) or not any( review.wait.current_for(goal, owner.incarnation.require_recorded()) for goal in owner.goal.active() @@ -278,26 +279,24 @@ def recover_closed_goal_wait(self, name: str) -> tuple[str, ...]: if not closed: return () # A failed read is a diagnostic, never evidence of silence. - for through_seq in review.unanswered(self.bus.bus): - with self.registry.store.locked(): - if not review.current_for(self): - return () - snapshot = self.registry.snapshot() - if self.agents.is_working(owner.incarnation, snapshot): - return () - closed = GoalWaits.closed_wait_group( - owner.name, review.wait.targets, self.waits.read(), snapshot, self.agents, - ) - if not closed: - return () - names = ", ".join(f"@{member}" for member in closed) - note = ( - f"Dependency wait group ({names}) has no independent worker. " - "Standby was released; inspect dependencies and continue useful work." - ) - if self._release_reviewed_wait(review, through_seq, note): - return closed - return () + with review.publishing_unanswered(self) as unchanged: + if not unchanged: + return () + snapshot = self.registry.snapshot() + owner, = snapshot.threads_for(review.owner.incarnation.require_recorded()) + if self.agents.is_working(owner.incarnation, snapshot): + return () + closed = GoalWaits.closed_wait_group( + owner.name, review.wait.targets, self.waits.read(), snapshot, self.agents, + ) + if not closed: + return () + names = ", ".join(f"@{member}" for member in closed) + note = ( + f"Dependency wait group ({names}) has no independent worker. " + "Standby was released; inspect dependencies and continue useful work." + ) + return closed if self._release_reviewed_wait(review, note) else () def release_waits_after_terminal_turn(self, fence: TurnCompletion) -> tuple[str, ...]: """A completed turn may release silence observed outside the registry lock.""" @@ -313,47 +312,47 @@ def release_waits_after_terminal_turn(self, fence: TurnCompletion) -> tuple[str, review = GoalWaitReview.capture(self, owner.name) except UnregisteredThreadError: continue - if review.wait == wait and self._release_finished_wait(review, fence): - released.append(review.owner.name) + if review.wait == wait: + released.extend(self._release_finished_wait(review, fence)) return tuple(released) - def _release_finished_wait(self, review: GoalWaitReview, fence: TurnCompletion) -> bool: + def _release_finished_wait(self, review: GoalWaitReview, fence: TurnCompletion) -> tuple[str, ...]: wait, owner = review.wait, review.owner if not any(wait.current_for(goal, owner.incarnation.require_recorded()) for goal in owner.goal.active()): - return False - if any(self.agents.is_working(target.incarnation, review.registry.snapshot) + return () + if any(self.agents.is_working(target.incarnation, review.snapshot) for target in wait.targets): - return False - for through_seq in review.unanswered(self.bus.bus): - with self.registry.store.locked(): - if not review.current_for(self): - return False - current = self.registry.snapshot() - sources = fence.idle_sources(current) - if not sources or any(self.agents.is_working(target.incarnation, current) - for target in wait.targets): - return False - actual_source, = sources - note = ( - f"Declared dependency @{actual_source.name} finished without a qualifying direct " - "reply, and no declared dependency has an active turn. " - "Standby was released; inspect messages and UNKNOWN inputs, then " - "continue independent work or redelegate. No input was replayed." - ) - return self._release_reviewed_wait(review, through_seq, note) - return False + return () + with review.publishing_unanswered(self) as unchanged: + if not unchanged: + return () + current = self.registry.snapshot() + sources = fence.idle_sources(current) + if not sources or any(self.agents.is_working(target.incarnation, current) + for target in wait.targets): + return () + actual_source, = sources + note = ( + f"Declared dependency @{actual_source.name} finished without a qualifying direct " + "reply, and no declared dependency has an active turn. " + "Standby was released; inspect messages and UNKNOWN inputs, then " + "continue independent work or redelegate. No input was replayed." + ) + if self._release_reviewed_wait(review, note): + actual_owner, = current.threads_for(owner.incarnation.require_recorded()) + return (actual_owner.name,) + return () - def _release_reviewed_wait(self, review: GoalWaitReview, through_seq: int, note: str) -> bool: - """Caller holds the registry; a new append invalidates the silence just read.""" - if self.bus.bus.revision() != through_seq: - return False - owner = review.owner + def _release_reviewed_wait(self, review: GoalWaitReview, note: str) -> bool: + """Publish a validated silent wait while its review owns the registry lock.""" + snapshot = self.registry.snapshot() + owner, = snapshot.threads_for(review.owner.incarnation.require_recorded()) goal = owner.goal.require_active() progress = f"{goal.progress}\n\n{note}" if goal.progress else note self.registry.register( replace(owner, goal=replace(goal, progress=progress, revision=goal.revision + 1)), - review.registry.snapshot.status(owner.name), + snapshot.status(owner.name), ) return review.wait.clear(self.waits) diff --git a/src/agent_comms/goal_waits.py b/src/agent_comms/goal_waits.py index 98e82c33ad..cc46d7fb92 100644 --- a/src/agent_comms/goal_waits.py +++ b/src/agent_comms/goal_waits.py @@ -3,7 +3,8 @@ from __future__ import annotations from abc import ABC, abstractmethod -from collections.abc import Iterable, Sequence +from collections.abc import Iterable, Iterator, Sequence +from contextlib import closing, contextmanager from dataclasses import dataclass, field from pathlib import Path from typing import TYPE_CHECKING, Annotated, ClassVar @@ -17,7 +18,6 @@ from .goals import Goal, GoalDeclaration from .input_attempt import StoredInput, UnresolvedInput from .locked_store import WireWakeFile -from .private_path import FileRevision from .store_files import SharedLock from .thread_identity import RecordedThreadIncarnation from .thread_owned_state import ThreadOwnedState @@ -29,7 +29,6 @@ from .registry_document import RegistrySnapshot from .threads import Thread from .goal_management import Goals - from .registry_store import RegistryRevision class GoalWait(ABC): @@ -257,36 +256,69 @@ class StandbyReview: class GoalWaitReview: """The existing registry, turn and wait observed before reading reply history.""" - registry: RegistryRevision + snapshot: RegistrySnapshot owner: Thread wait: GoalWait turn: TurnLease - @classmethod - def capture(cls, goals: Goals, name: str) -> GoalWaitReview: + @staticmethod + def capture(goals: Goals, name: str) -> GoalWaitReview: with goals.registry.store.locked(mode=SharedLock()): - registry = goals.registry.store.read_revision_unlocked() - owner = registry.snapshot.require(name) - return cls(registry, owner, goals.waits.resolve(owner.goal), owner.turn_lease) + snapshot = goals.registry.store.read_snapshot_unlocked() + owner = snapshot.require(name) + return GoalWaitReview(snapshot, owner, goals.waits.resolve(owner.goal), owner.turn_lease) 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( + self._same_thread(snapshot, owner, self.owner) + and owner.goal == self.owner.goal + 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()) + ) + + def _same_thread(self, snapshot: RegistrySnapshot, actual: Thread, expected: Thread) -> bool: + """Ignore display metadata, but retain the original process and input/turn epochs.""" + before = self.snapshot return ( - FileRevision.capture(goals.registry.store.path) == self.registry.revision - and self.owner.turn_lease == self.turn - and goals.waits.resolve(self.owner.goal) == self.wait + actual.same_launch(expected) + and actual.turn_generation == expected.turn_generation + and actual.turn_record == expected.turn_record + and snapshot.owner_identity(actual.name) == before.owner_identity(expected.name) + and snapshot.claim_generation(actual.name) == before.claim_generation(expected.name) + and snapshot.status(actual.name) == before.status(expected.name) ) - def require_current(self, goals: Goals) -> None: - if not self.current_for(goals): + def require_current(self, goals: Goals, owner: Thread) -> None: + if owner.incarnation != self.owner.incarnation or not self.current_for(goals): raise ValueError("Goal, dependency or wait changed during reply review; inspect it again.") - def unanswered(self, bus: BusReader) -> tuple[int, ...]: - """The captured bus cut with no qualifying reply, or no such cut.""" - with bus.reading() as source: - if self.wait.has_reply(source, self.owner.incarnation.require_recorded()): - return () - return (source.revision(),) + @contextmanager + 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() + while True: + with goals.bus.bus.reading() as source: + replied = self.wait.has_reply(source, owner) + through_seq = source.revision() + if replied: + yield False + return + with goals.registry.store.locked(): + if not self.current_for(goals): + yield False + return + # The existing indexed sequence query reads at most one metadata row. + # Any new candidate needs another unlocked policy read, not a stale release. + with closing(goals.bus.bus._received_sequences( + owner.lookup, after=through_seq, limit=1, + )) as positions: + if next(positions, None) is not None: + continue + yield True + return @property def goal_id(self) -> str: @@ -294,7 +326,7 @@ def goal_id(self) -> str: @property def owners(self) -> frozenset[str]: - return self.registry.snapshot.names_for(self.owner.name) + return self.snapshot.names_for(self.owner.name) @dataclass(frozen=True, slots=True) @@ -306,6 +338,16 @@ class GoalInputReview(GoalWaitReview): unknown: tuple[StoredInput, ...] eligible_keys: frozenset[str] + def current_for(self, goals: Goals) -> bool: + if not super(GoalInputReview, self).current_for(goals): + return False + snapshot = goals.registry.snapshot() + before = self.snapshot + return all(any( + self._same_thread(snapshot, actual, before.require(target.incarnation.name)) + for actual in snapshot.threads_for(target.incarnation) + ) for target in self.targets) + def standby_review(self) -> StandbyReview: eligible: list[UnresolvedInput] = [] reviewed: list[UnresolvedInput] = [] diff --git a/tests/test_goal_reply_lock_scope.py b/tests/test_goal_reply_lock_scope.py index 6cb67ce0c9..d4dee4b5a1 100644 --- a/tests/test_goal_reply_lock_scope.py +++ b/tests/test_goal_reply_lock_scope.py @@ -4,6 +4,7 @@ import subprocess import sys from contextlib import contextmanager +from dataclasses import replace import pytest @@ -21,8 +22,12 @@ ) from agent_comms.goal_waits import GoalWaitReview, RecordedGoalWait from agent_comms.input_disposition import InputDispositions -from agent_comms.native_input_owner import RegistryOwner +from agent_comms.coordination_errors import StaleFence +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.tool_caller import ThreadCaller +from agent_comms.tools import CommsGoalTool from goal_owner_fixture import canonical_goal_wire @@ -60,6 +65,23 @@ def observing(self): db.set_trace_callback(None) +class AfterReadBus(Bus): + """Change real state once the real query snapshot closes, before publication.""" + + def __init__(self, path, change): + super().__init__(path) + self.change = change + self.changed = False + + @contextmanager + def reading(self): + with super().reading() as source: + yield source + if not self.changed: + self.changed = True + self.change() + + @pytest.fixture def reviewing(tmp_path): comms = canonical_goal_wire(tmp_path / "wire") @@ -87,7 +109,10 @@ def standby(comms, reviewed_inputs=()): def test_standby_reads_replies_without_holding_registry(probed): - probed.goals.update_goal("owner", standby(probed)) + CommsGoalTool.invoke(probed, { + "goal_id": probed.registry.require("owner").goal.require().id, + "status": "standby", "progress": "Waiting for the child", "wait_for": ["child"], + }, caller=ThreadCaller("owner")) assert probed.bus.bus.checks == [0] @@ -112,10 +137,7 @@ def test_completed_reply_is_decoded_without_holding_registry(probed): probed.goals.update_goal("owner", standby(probed)) reply = probed.messaging.send("child", "owner", "Finished").message probed.bus.bus.checks.clear() - owner = probed.registry.require("owner") - witness = RegistryOwner( - thread=owner, claimed_generation=probed.registry.snapshot().claim_generation(owner.name), - ) + witness = probed.agents.begin_turn("owner", "handling-reply") assert probed.goals.consume_reply_wait(witness, reply.reference) assert probed.bus.bus.checks == [0] @@ -135,7 +157,9 @@ def test_reply_appended_between_review_and_publication_is_still_eligible(reviewi assert review.reviewed_inputs == (f"bus:{reply.seq}",) -@pytest.mark.parametrize("change", ("goal", "wait", "turn", "dependency")) +@pytest.mark.parametrize("change", ( + "goal", "wait", "turn", "dependency_turn", "dependency", "owner", "input", +)) def test_stale_review_never_writes_inputs_or_replaces_newer_wait(reviewing, change): reply = reviewing.messaging.send("child", "owner", "Review this reply").message prepared = standby(reviewing, (f"bus:{reply.seq}",)).prepare( @@ -155,32 +179,122 @@ def test_stale_review_never_writes_inputs_or_replaces_newer_wait(reviewing, chan )) elif change == "turn": reviewing.agents.begin_turn("owner", "new-owner-turn") + elif change == "dependency_turn": + reviewing.agents.finish_turn(reviewing.registry.require("child").require_turn_lease()) + reviewing.agents.begin_turn("child", "new-child-turn") + elif change == "input": + InputDispositions(reviewing.root).review_for_goal( + prepared.reviewed_inputs, observed=prepared.review.unknown, + owners=prepared.review.owners, goal_id=goal.id, + goal_revision=goal.revision + 1, turn_id="", + ) else: - reviewing.registry.unregister("child") - reviewing.registry.remove("child") + name = "owner" if change == "owner" else "child" + reviewing.registry.unregister(name) + reviewing.registry.remove(name) reviewing.registry.declare(reviewing.registry.new_thread( - "child", frozenset(), owner.worktree, + name, frozenset(), owner.worktree, process_identity=ProcessIdentity.capture(os.getpid()), )) inputs = InputDispositions(reviewing.root) - before = inputs.path.read_bytes() if inputs.path.exists() else None + before = inputs.read(prepared.reviewed_inputs) wait_before = reviewing.goals.goal_wait("owner") + goal_before = reviewing.registry.require("owner").goal with reviewing.registry.store.locked(), pytest.raises(ValueError): prepared.apply(GoalActionContext( reviewing.goals, reviewing.registry.require("owner"), RuntimeInvocable, )) - assert (inputs.path.read_bytes() if inputs.path.exists() else None) == before + assert inputs.read(prepared.reviewed_inputs) == before assert reviewing.goals.goal_wait("owner") == wait_before + assert reviewing.registry.require("owner").goal == goal_before -def test_late_reply_invalidates_captured_silence(reviewing): +@pytest.mark.parametrize("role,released", ((ConversationRole(), ()), (NoticeRole(), ("owner",)))) +def test_new_received_candidate_is_read_before_releasing_silence(reviewing, role, released): reviewing.goals.update_goal("owner", standby(reviewing)) - reviewing.agents.finish_turn(reviewing.registry.require("child").require_turn_lease()) - captured = GoalWaitReview.capture(reviewing.goals, "owner") - cut, = captured.unanswered(reviewing.bus.bus) - reviewing.messaging.send("child", "owner", "A reply after the silence check") + fence = reviewing.agents.finish_turn(reviewing.registry.require("child").require_turn_lease()) + wait = reviewing.goals.goal_wait("owner") + reviewing.bus.bus = AfterReadBus(reviewing.bus.bus.path, lambda: reviewing.messaging.send( + "child", "owner", "A message after the silence check", role=role, + )) + assert reviewing.goals.release_waits_after_terminal_turn(fence) == released + assert (reviewing.goals.goal_wait("owner") != wait) == bool(released) + + +def test_other_recipient_traffic_does_not_strand_terminal_wait(reviewing): + reviewing.registry.declare(reviewing.registry.new_thread("other", frozenset(), str(reviewing.root))) + reviewing.goals.update_goal("owner", standby(reviewing)) + fence = reviewing.agents.finish_turn(reviewing.registry.require("child").require_turn_lease()) + reviewing.bus.bus = AfterReadBus(reviewing.bus.bus.path, lambda: reviewing.messaging.send( + "child", "other", "This has nothing to do with the waiting owner", + )) + assert reviewing.goals.release_waits_after_terminal_turn(fence) == ("owner",) + + +def test_terminal_reply_read_cannot_clear_a_newer_wait(reviewing): + reviewing.goals.update_goal("owner", standby(reviewing)) + fence = reviewing.agents.finish_turn(reviewing.registry.require("child").require_turn_lease()) + goal = reviewing.registry.require("owner").goal + newer = replace(reviewing.goals.goal_wait("owner"), wait_id="newer-wait") + reviewing.bus.bus = AfterReadBus( + reviewing.bus.bus.path, lambda: reviewing.goals.waits.record(newer), + ) + assert reviewing.goals.release_waits_after_terminal_turn(fence) == () + assert reviewing.goals.goal_wait("owner") == newer + assert reviewing.registry.require("owner").goal == goal + + +@pytest.mark.parametrize("change", ("wait", "turn")) +def test_completed_reply_rechecks_its_wait_and_original_turn(reviewing, change): + reviewing.goals.update_goal("owner", standby(reviewing)) + reply = reviewing.messaging.send("child", "owner", "The actual completed reply").message + witness = reviewing.agents.begin_turn("owner", "original-handling-turn") + wait = reviewing.goals.goal_wait("owner") + newer = replace(wait, wait_id="newer-wait") + + def replace_state(): + if change == "wait": + reviewing.goals.waits.record(newer) + else: + reviewing.agents.finish_turn(witness.thread.require_turn_lease()) + reviewing.agents.begin_turn("owner", "a-later-turn") + + reviewing.bus.bus = AfterReadBus(reviewing.bus.bus.path, replace_state) + if change == "wait": + assert not reviewing.goals.consume_reply_wait(witness, reply.reference) + assert reviewing.goals.goal_wait("owner") == newer + else: + with pytest.raises(StaleFence): + reviewing.goals.consume_reply_wait(witness, reply.reference) + assert reviewing.goals.goal_wait("owner") == wait + + +@pytest.mark.parametrize("rename", (False, True)) +def test_display_edit_during_standby_review_is_preserved(reviewing, rename): + prepared = standby(reviewing).prepare(reviewing.goals, "owner", RuntimeInvocable) + title = GivenThreadText("A title changed while replies were read") + reviewing.registry.register(replace(reviewing.registry.require("owner"), title=title)) + if rename: + reviewing.registry.rename("owner", "renamed-owner") with reviewing.registry.store.locked(): - assert captured.current_for(reviewing.goals) - assert not reviewing.goals._release_reviewed_wait(captured, cut, "stale silence") - assert reviewing.goals.goal_wait("owner") == captured.wait - assert reviewing.registry.require("owner").goal == captured.owner.goal + prepared.apply(GoalActionContext( + reviewing.goals, reviewing.registry.require("owner"), RuntimeInvocable, + )) + assert reviewing.registry.require("owner").title == title + assert reviewing.goals.goal_wait("owner").goal_id == prepared.review.goal_id + + +@pytest.mark.parametrize("rename", (False, True)) +def test_display_edit_during_terminal_review_does_not_strand_wait(reviewing, rename): + reviewing.goals.update_goal("owner", standby(reviewing)) + fence = reviewing.agents.finish_turn(reviewing.registry.require("child").require_turn_lease()) + captured = GoalWaitReview.capture(reviewing.goals, "owner") + title = GivenThreadText("Keep this concurrent title") + reviewing.registry.register(replace(reviewing.registry.require("owner"), title=title)) + if rename: + reviewing.registry.rename("owner", "renamed-owner") + assert reviewing.goals._release_finished_wait(captured, fence) == ( + "renamed-owner" if rename else "owner", + ) + assert reviewing.registry.require("owner").title == title + assert reviewing.goals.goal_wait("owner") != captured.wait From fb6c579fd0f0ed03197a0fca81ad112b114f2abb Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Fri, 23 Oct 2026 03:00:40 -0400 Subject: [PATCH 3/5] Keep turn and registry authority comparisons with their owners --- src/agent_comms/goal_management.py | 8 +++++--- src/agent_comms/goal_waits.py | 24 ++++++++++++------------ src/agent_comms/private_send_rules.py | 8 +------- src/agent_comms/registry_document.py | 11 +++++++++++ src/agent_comms/threads.py | 11 +++++++++++ 5 files changed, 40 insertions(+), 22 deletions(-) diff --git a/src/agent_comms/goal_management.py b/src/agent_comms/goal_management.py index 6067f5592f..ce011b5088 100644 --- a/src/agent_comms/goal_management.py +++ b/src/agent_comms/goal_management.py @@ -26,8 +26,8 @@ from .goal_history import GoalHistoryEntry from .goal_waits import ( GoalInputReview, - GoalWaitReview, GoalWait, + GoalWaitReview, GoalWaits, RecordedGoalWait, StandbyReview, @@ -53,7 +53,7 @@ GoalWaitTarget, IdleGoalExecution, ) -from .goals import Goal, GoalDeclaration +from .goals import GoalDeclaration from .input_attempt import ( UNOBSERVED_DELIVERY_SCOPE, InputDeliveryScope, @@ -316,7 +316,9 @@ def release_waits_after_terminal_turn(self, fence: TurnCompletion) -> tuple[str, released.extend(self._release_finished_wait(review, fence)) return tuple(released) - def _release_finished_wait(self, review: GoalWaitReview, fence: TurnCompletion) -> tuple[str, ...]: + def _release_finished_wait( + self, review: GoalWaitReview, fence: TurnCompletion, + ) -> tuple[str, ...]: wait, owner = review.wait, review.owner if not any(wait.current_for(goal, owner.incarnation.require_recorded()) for goal in owner.goal.active()): diff --git a/src/agent_comms/goal_waits.py b/src/agent_comms/goal_waits.py index cc46d7fb92..548ea77424 100644 --- a/src/agent_comms/goal_waits.py +++ b/src/agent_comms/goal_waits.py @@ -26,9 +26,9 @@ if TYPE_CHECKING: from .agent_activity import AgentActivity + from .goal_management import Goals from .registry_document import RegistrySnapshot from .threads import Thread - from .goal_management import Goals class GoalWait(ABC): @@ -266,34 +266,34 @@ def capture(goals: Goals, name: str) -> GoalWaitReview: with goals.registry.store.locked(mode=SharedLock()): snapshot = goals.registry.store.read_snapshot_unlocked() owner = snapshot.require(name) - return GoalWaitReview(snapshot, owner, goals.waits.resolve(owner.goal), owner.turn_lease) + return GoalWaitReview( + snapshot, owner, goals.waits.resolve(owner.goal), owner.turn_lease, + ) 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( - self._same_thread(snapshot, owner, self.owner) - and owner.goal == self.owner.goal + owner.goal == self.owner.goal 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 _same_thread(self, snapshot: RegistrySnapshot, actual: Thread, expected: Thread) -> bool: """Ignore display metadata, but retain the original process and input/turn epochs.""" - before = self.snapshot return ( actual.same_launch(expected) - and actual.turn_generation == expected.turn_generation - and actual.turn_record == expected.turn_record - and snapshot.owner_identity(actual.name) == before.owner_identity(expected.name) - and snapshot.claim_generation(actual.name) == before.claim_generation(expected.name) - and snapshot.status(actual.name) == before.status(expected.name) + and actual.same_turn(expected) + and snapshot.same_authority(actual.name, self.snapshot, expected.name) ) def require_current(self, goals: Goals, owner: Thread) -> None: if owner.incarnation != self.owner.incarnation or not self.current_for(goals): - raise ValueError("Goal, dependency or wait changed during reply review; inspect it again.") + raise ValueError( + "Goal, dependency or wait changed during reply review; inspect it again." + ) @contextmanager def publishing_unanswered(self, goals: Goals) -> Iterator[bool]: @@ -315,7 +315,7 @@ def publishing_unanswered(self, goals: Goals) -> Iterator[bool]: with closing(goals.bus.bus._received_sequences( owner.lookup, after=through_seq, limit=1, )) as positions: - if next(positions, None) is not None: + if any(positions): continue yield True return diff --git a/src/agent_comms/private_send_rules.py b/src/agent_comms/private_send_rules.py index 890f4c8c87..0e3a4ccac1 100644 --- a/src/agent_comms/private_send_rules.py +++ b/src/agent_comms/private_send_rules.py @@ -167,13 +167,7 @@ class RegistryTurnRule(ReservationRule): explanation = "The exact captured active turn changed." def violated(self, check: RegistryClaimCheck) -> bool: - # Compare the captured references, not two fresh reads of mutable records. - # Revoking a claim makes both reads return None; it does not make the - # captured turn current again. The generation also fences idle captures. - return ( - check.actual.turn_record != check.expected.turn_record - or check.actual.turn_generation != check.expected.turn_generation - ) + return not check.actual.same_turn(check.expected) class RegistryGoalRule(ReservationRule): diff --git a/src/agent_comms/registry_document.py b/src/agent_comms/registry_document.py index beca3d3e0f..1adee52d6c 100644 --- a/src/agent_comms/registry_document.py +++ b/src/agent_comms/registry_document.py @@ -875,6 +875,17 @@ def claimed_identity(self, name: str) -> ClaimedIdentity: canonical = self.canonical_name(name) return ClaimedIdentity(self.threads[canonical].incarnation, self.claim_generation(canonical)) + def same_authority(self, name: str, other: RegistrySnapshot, other_name: str) -> bool: + """One participant retains its owner, input claim and registered presence. + + Titles, aliases and unrelated rows do not allocate or revoke authority. + """ + return ( + self.owner_identity(name) == other.owner_identity(other_name) + and self.claimed_identity(name) == other.claimed_identity(other_name) + and self.status(name) == other.status(other_name) + ) + def owner_binding(self, name: str) -> ThreadOwnerBinding: canonical = self.canonical_name(name) thread = self.threads[canonical] diff --git a/src/agent_comms/threads.py b/src/agent_comms/threads.py index 6e6820fab4..96073b5874 100644 --- a/src/agent_comms/threads.py +++ b/src/agent_comms/threads.py @@ -398,6 +398,17 @@ def launch(thread: Thread): thread.worktree, thread.execution, thread.model, thread.thinking_level) return launch(self) == launch(other) + def same_turn(self, other: Thread) -> bool: + """Compare captured references and epochs, including captures made while idle. + + Reading the mutable turn files again could make two revoked leases look + equal. The saved references and registry generations retain their fence. + """ + return ( + self.turn_record == other.turn_record + and self.turn_generation == other.turn_generation + ) + def without_turn(self) -> Thread: """Drop the reference; the registry transaction owns revoking the record.""" return replace(self, turn_record=NoTurnRecord()) From 695e4f23909d9c414d9069803c8c19afb4cd41ea Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Fri, 23 Oct 2026 03:09:17 -0400 Subject: [PATCH 4/5] Preserve reviewed incarnation and use owned reply and authority queries --- src/agent_comms/bus.py | 5 + src/agent_comms/goal_actions.py | 2 +- src/agent_comms/goal_management.py | 13 ++- src/agent_comms/goal_waits.py | 21 ++-- src/agent_comms/owner_restart.py | 30 ++--- src/agent_comms/registry_document.py | 7 +- src/agent_comms/session_revision.py | 2 +- src/agent_comms/thread_identity.py | 8 ++ tests/test_goal_reply_lock_scope.py | 169 ++++++++++++++++----------- tests/test_owner_restart_handoff.py | 7 +- tests/test_reservation_rules.py | 28 +++-- 11 files changed, 176 insertions(+), 116 deletions(-) diff --git a/src/agent_comms/bus.py b/src/agent_comms/bus.py index 25f0c9b891..12b91ceb3a 100644 --- a/src/agent_comms/bus.py +++ b/src/agent_comms/bus.py @@ -386,6 +386,11 @@ class Bus(BusReader, ReconciledFile, WalStore, CarriedWriter, BusFile, Watched): def __init__(self, path: Path): super().__init__(path) + def received_after(self, lookup: str, after: int) -> bool: + """Whether this recipient has a newer candidate; read one indexed metadata row.""" + with closing(self._received_sequences(lookup, after=after, limit=1)) as positions: + return any(positions) + @classmethod def install(cls, path: Path, *, history: Iterable[Message] = ()) -> Self: """Install declared schemas and commit optional display-only history atomically. diff --git a/src/agent_comms/goal_actions.py b/src/agent_comms/goal_actions.py index a2835e8f18..38c3bb83eb 100644 --- a/src/agent_comms/goal_actions.py +++ b/src/agent_comms/goal_actions.py @@ -270,7 +270,7 @@ def next_state(self, goal: Goal, ctx: GoalActionContext) -> GoalState: def prepare(self, goals: Goals, name: str, actor: type) -> GoalAction[Goal]: ctx = GoalActionContext(goals, goals.registry.require(name), actor) self.expect.check(ctx) - review = goals._goal_input_review(name, ctx.thread.goal.require().id, self.wait_for) + review = goals._goal_input_review(ctx.thread, ctx.thread.goal.require().id, self.wait_for) self.expect.check(GoalActionContext(goals, review.owner, actor)) return PreparedStandbyGoalAction( expect=replace(self.expect, expected_goal=review.owner.goal), diff --git a/src/agent_comms/goal_management.py b/src/agent_comms/goal_management.py index ce011b5088..6c8f7ade16 100644 --- a/src/agent_comms/goal_management.py +++ b/src/agent_comms/goal_management.py @@ -168,13 +168,13 @@ def dismiss_historical_inputs( def goal_input_review(self, name: str, goal_id: str, wait_for: Sequence[str]) -> StandbyReview: """Project exact review eligibility for one current goal and dependency set.""" - review = self._goal_input_review(name, goal_id, wait_for) + review = self._goal_input_review(self.registry.require(name), goal_id, wait_for) return review.standby_review() def _goal_input_review( - self, name: str, goal_id: str, wait_for: Sequence[str] + self, thread: Thread, goal_id: str, wait_for: Sequence[str] ) -> GoalInputReview: - captured = GoalWaitReview.capture(self, name) + captured = GoalWaitReview.capture(self, thread.incarnation.require_recorded()) thread = captured.owner snapshot = captured.snapshot goal = thread.goal.require().require_id(goal_id).require_active() @@ -263,7 +263,8 @@ def goal_wait(self, name: str) -> GoalWait: def recover_closed_goal_wait(self, name: str) -> tuple[str, ...]: """Read silence outside the registry, then release only the observed wait.""" try: - review = GoalWaitReview.capture(self, name) + incarnation = self.registry.require(name).incarnation.require_recorded() + review = GoalWaitReview.capture(self, incarnation) except UnregisteredThreadError: return () owner = review.owner @@ -309,7 +310,7 @@ def release_waits_after_terminal_turn(self, fence: TurnCompletion) -> tuple[str, for wait in self.waits.targeting(source.incarnation.require_recorded()): for owner in snapshot.threads_for(wait.owner): try: - review = GoalWaitReview.capture(self, owner.name) + review = GoalWaitReview.capture(self, owner.incarnation.require_recorded()) except UnregisteredThreadError: continue if review.wait == wait: @@ -403,7 +404,7 @@ def consume_reply_wait(self, owner: RegistryOwner, reference: MessageReference) """Read the exact completed delivery before checking its original owner fence.""" with self.registry.store.locked(mode=SharedLock()): owner.require_registry(self.registry) - review = GoalWaitReview.capture(self, owner.thread.name) + 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()): diff --git a/src/agent_comms/goal_waits.py b/src/agent_comms/goal_waits.py index 548ea77424..f0dda22670 100644 --- a/src/agent_comms/goal_waits.py +++ b/src/agent_comms/goal_waits.py @@ -4,7 +4,7 @@ from abc import ABC, abstractmethod from collections.abc import Iterable, Iterator, Sequence -from contextlib import closing, contextmanager +from contextlib import contextmanager from dataclasses import dataclass, field from pathlib import Path from typing import TYPE_CHECKING, Annotated, ClassVar @@ -13,6 +13,7 @@ from .bus_publication import Delivered from .bus_tables import DirectReplySource from .declared_family import DeclaredFamily +from .errors import UnregisteredThreadError from .field_codec import NullMember from .goal_presentation import GoalExecution, GoalExecutionSource, GoalWaitTarget, StandbyExecution from .goals import Goal, GoalDeclaration @@ -262,13 +263,14 @@ class GoalWaitReview: turn: TurnLease @staticmethod - def capture(goals: Goals, name: str) -> GoalWaitReview: + def capture(goals: Goals, incarnation: RecordedThreadIncarnation) -> GoalWaitReview: with goals.registry.store.locked(mode=SharedLock()): snapshot = goals.registry.store.read_snapshot_unlocked() - owner = snapshot.require(name) - return GoalWaitReview( - snapshot, owner, goals.waits.resolve(owner.goal), owner.turn_lease, - ) + for owner in snapshot.threads_for(incarnation): + return GoalWaitReview( + snapshot, owner, goals.waits.resolve(owner.goal), owner.turn_lease, + ) + raise UnregisteredThreadError("The reviewed thread incarnation was removed.") def current_for(self, goals: Goals) -> bool: """Caller holds the publication lock; no reply decoding belongs in this check.""" @@ -312,11 +314,8 @@ def publishing_unanswered(self, goals: Goals) -> Iterator[bool]: return # The existing indexed sequence query reads at most one metadata row. # Any new candidate needs another unlocked policy read, not a stale release. - with closing(goals.bus.bus._received_sequences( - owner.lookup, after=through_seq, limit=1, - )) as positions: - if any(positions): - continue + if goals.bus.bus.received_after(owner.lookup, through_seq): + continue yield True return diff --git a/src/agent_comms/owner_restart.py b/src/agent_comms/owner_restart.py index 5c2686180d..f9f6c3a140 100644 --- a/src/agent_comms/owner_restart.py +++ b/src/agent_comms/owner_restart.py @@ -9,19 +9,23 @@ import os from contextlib import ExitStack from dataclasses import dataclass, field, replace -from typing import Annotated, TYPE_CHECKING - -from agent_comms import field_codec +from typing import TYPE_CHECKING from .coordination_errors import CoordinationError from .errors import RelationViolationError from .launch_runtime import LaunchRuntime, RestartEnvironment, RestoredRuntime +from .native_arguments import ( + ConfiguredBinary, + GivenArguments, + InheritedArguments, + LaunchArguments, + LaunchBinary, +) from .owner_launch import RetainedOwnerLaunch from .owner_lifecycle import OwnerRestartSelection from .restart_refusals import OwnerSelectionChangedRefusal from .restart_scope import AllOwners, RestartScope -from .thread_identity import ClaimedIdentity, OwnerIdentity -from .native_arguments import ConfiguredBinary, GivenArguments, InheritedArguments, LaunchArguments, LaunchBinary +from .thread_identity import ThreadAuthority if TYPE_CHECKING: from .owner_cutover import OwnerCutover @@ -58,8 +62,7 @@ def environment(self, lifecycle: OwnerLifecycle) -> RestartEnvironment: class RetiredOwnerLaunch: """Original stopped registry observation plus acquired launch resources.""" - owner: OwnerIdentity - claim_scope: Annotated[ClaimedIdentity, field_codec.WireName("admission")] + authority: ThreadAuthority launch: RetainedOwnerLaunch @classmethod @@ -67,22 +70,19 @@ def capture_retired(cls, snapshot: RegistrySnapshot, original: Thread, launch: RetainedOwnerLaunch) -> RetiredOwnerLaunch: """Capture the original allocations after stopping, without permitting live work.""" original.require_local_process(launch.process) - owner = cls(snapshot.owner_identity(original.name), - snapshot.claimed_identity(original.name), launch) - if owner.owner.incarnation != original.incarnation: + owner = cls(snapshot.authority(original.name), launch) + if owner.authority.owner.incarnation != original.incarnation: raise RelationViolationError('Retired owner incarnation changed') owner.require_current(snapshot) return owner @property def name(self) -> str: - return self.owner.incarnation.name + return self.authority.owner.incarnation.name def require_current(self, snapshot: RegistrySnapshot) -> Thread: - if snapshot.owner_identity(self.name) != self.owner: - raise RelationViolationError('Retired owner incarnation/generation changed') - if snapshot.claimed_identity(self.name) != self.claim_scope: - raise RelationViolationError('Retired owner claim_scope changed') + if snapshot.authority(self.name) != self.authority: + raise RelationViolationError('Retired owner authority changed') thread = snapshot.threads[self.name] thread.require_local_process(self.launch.process) thread.turn_record.read().require_idle() diff --git a/src/agent_comms/registry_document.py b/src/agent_comms/registry_document.py index 1adee52d6c..2b3fb606db 100644 --- a/src/agent_comms/registry_document.py +++ b/src/agent_comms/registry_document.py @@ -44,6 +44,7 @@ GenerationCounter, OwnerIdentity, RecordedThreadIncarnation, + ThreadAuthority, TurnIdentity, ) from .thread_presentation import ThreadOwnerBinding @@ -881,11 +882,13 @@ def same_authority(self, name: str, other: RegistrySnapshot, other_name: str) -> Titles, aliases and unrelated rows do not allocate or revoke authority. """ return ( - self.owner_identity(name) == other.owner_identity(other_name) - and self.claimed_identity(name) == other.claimed_identity(other_name) + self.authority(name) == other.authority(other_name) and self.status(name) == other.status(other_name) ) + def authority(self, name: str) -> ThreadAuthority: + return ThreadAuthority(self.owner_identity(name), self.claimed_identity(name)) + def owner_binding(self, name: str) -> ThreadOwnerBinding: canonical = self.canonical_name(name) thread = self.threads[canonical] diff --git a/src/agent_comms/session_revision.py b/src/agent_comms/session_revision.py index a366ffb8a9..e97f865cf0 100644 --- a/src/agent_comms/session_revision.py +++ b/src/agent_comms/session_revision.py @@ -66,7 +66,7 @@ def observe(cls, files: NativeSessionFiles) -> SessionObservation: except FileNotFoundError: proof = MissingInputProofRevision() return cls(native, proof) - except (FileNotFoundError, PermissionError) as error: + except OSError as error: # Unreadable proof is not a missing proof. Deny source acquisition. return UnavailableSessionObservation(error) diff --git a/src/agent_comms/thread_identity.py b/src/agent_comms/thread_identity.py index 71af5ff476..fb1cb4af4f 100644 --- a/src/agent_comms/thread_identity.py +++ b/src/agent_comms/thread_identity.py @@ -416,6 +416,14 @@ def includes(self, previous: ClaimedIdentity) -> bool: return self.incarnation == previous.incarnation and self.claimed_generation >= previous.claimed_generation +@dataclass(frozen=True, slots=True) +class ThreadAuthority: + """One registry participant's owner lease and allocated input claim.""" + + owner: OwnerIdentity + claim: ClaimedIdentity + + @dataclass(frozen=True, slots=True) class TurnIdentity: incarnation: ThreadIncarnation diff --git a/tests/test_goal_reply_lock_scope.py b/tests/test_goal_reply_lock_scope.py index d4dee4b5a1..d46708a1dd 100644 --- a/tests/test_goal_reply_lock_scope.py +++ b/tests/test_goal_reply_lock_scope.py @@ -1,10 +1,10 @@ """Reply reads leave the registry free, and publication keeps its observed state.""" +import fcntl import os -import subprocess -import sys from contextlib import contextmanager from dataclasses import replace +from functools import partial import pytest @@ -12,6 +12,7 @@ from agent_comms.bus_publication import Delivered from agent_comms.bus_tables import ReceivedSource from agent_comms.child_process import ProcessIdentity +from agent_comms.coordination_errors import StaleFence from agent_comms.goal_actions import ( EditGoalAction, GoalActionContext, @@ -22,7 +23,6 @@ ) from agent_comms.goal_waits import GoalWaitReview, RecordedGoalWait from agent_comms.input_disposition import InputDispositions -from agent_comms.coordination_errors import StaleFence from agent_comms.messages import ConversationRole, NoticeRole from agent_comms.store_files import store_lock_path from agent_comms.thread_text import GivenThreadText @@ -32,7 +32,7 @@ class RegistryProbeBus(Bus): - """A real SQLite read asks another process to take the real registry lock.""" + """A second open file description tries to take the real registry lock.""" def __init__(self, path, registry_path): super().__init__(path) @@ -44,16 +44,14 @@ def probe(self, statement): ReceivedSource.declared_name, Delivered.table.declared_name, )): return - checked = subprocess.run( - [sys.executable, "-c", ( - "import fcntl, sys\n" - "with open(sys.argv[1], 'rb') as lock:\n" - " try: fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)\n" - " except BlockingIOError: sys.exit(11)\n" - ), str(self.registry_lock)], - capture_output=True, text=True, timeout=5, - ) - self.checks.append(checked.returncode) + with self.registry_lock.open("rb") as lock: + try: + fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB) + except BlockingIOError: + self.checks.append(11) + else: + self.checks.append(0) + fcntl.flock(lock, fcntl.LOCK_UN) @contextmanager def observing(self): @@ -157,45 +155,61 @@ def test_reply_appended_between_review_and_publication_is_still_eligible(reviewi assert review.reviewed_inputs == (f"bus:{reply.seq}",) +def change_goal(comms, prepared): + comms.goals.update_goal("owner", EditGoalAction(text="A different objective")) + + +def replace_wait(comms, prepared): + owner = comms.registry.require("owner") + goal = owner.goal.require() + target = comms.registry.require("child") + comms.goals.waits.record(RecordedGoalWait( + goal.id, "newer-wait", goal.revision, prepared.review.through_seq, + prepared.review.targets, owner=owner.incarnation.require_recorded(), + target_turn_generations=(target.require_turn_lease().identity.generation,), + report_turn_id=None, report_turn_generation=None, + )) + + +def begin_owner_turn(comms, prepared): + comms.agents.begin_turn("owner", "new-owner-turn") + + +def begin_dependency_turn(comms, prepared): + comms.agents.finish_turn(comms.registry.require("child").require_turn_lease()) + comms.agents.begin_turn("child", "new-child-turn") + + +def review_inputs(comms, prepared): + goal = comms.registry.require("owner").goal.require() + InputDispositions(comms.root).review_for_goal( + prepared.reviewed_inputs, observed=prepared.review.unknown, + owners=prepared.review.owners, goal_id=goal.id, + goal_revision=goal.revision + 1, turn_id="", + ) + + +def replace_registration(comms, prepared, *, name): + original = comms.registry.require(name) + comms.registry.unregister(name) + comms.registry.remove(name) + comms.registry.declare(comms.registry.new_thread( + name, frozenset(), original.worktree, + process_identity=ProcessIdentity.capture(os.getpid()), + )) + + @pytest.mark.parametrize("change", ( - "goal", "wait", "turn", "dependency_turn", "dependency", "owner", "input", + change_goal, replace_wait, begin_owner_turn, begin_dependency_turn, review_inputs, + partial(replace_registration, name="child"), + partial(replace_registration, name="owner"), )) def test_stale_review_never_writes_inputs_or_replaces_newer_wait(reviewing, change): reply = reviewing.messaging.send("child", "owner", "Review this reply").message prepared = standby(reviewing, (f"bus:{reply.seq}",)).prepare( reviewing.goals, "owner", RuntimeInvocable, ) - owner = reviewing.registry.require("owner") - goal = owner.goal.require() - if change == "goal": - reviewing.goals.update_goal("owner", EditGoalAction(text="A different objective")) - elif change == "wait": - target = reviewing.registry.require("child") - reviewing.goals.waits.record(RecordedGoalWait( - goal.id, "newer-wait", goal.revision, reply.seq, prepared.review.targets, - owner=owner.incarnation.require_recorded(), - target_turn_generations=(target.require_turn_lease().identity.generation,), - report_turn_id=None, report_turn_generation=None, - )) - elif change == "turn": - reviewing.agents.begin_turn("owner", "new-owner-turn") - elif change == "dependency_turn": - reviewing.agents.finish_turn(reviewing.registry.require("child").require_turn_lease()) - reviewing.agents.begin_turn("child", "new-child-turn") - elif change == "input": - InputDispositions(reviewing.root).review_for_goal( - prepared.reviewed_inputs, observed=prepared.review.unknown, - owners=prepared.review.owners, goal_id=goal.id, - goal_revision=goal.revision + 1, turn_id="", - ) - else: - name = "owner" if change == "owner" else "child" - reviewing.registry.unregister(name) - reviewing.registry.remove(name) - reviewing.registry.declare(reviewing.registry.new_thread( - name, frozenset(), owner.worktree, - process_identity=ProcessIdentity.capture(os.getpid()), - )) + change(reviewing, prepared) inputs = InputDispositions(reviewing.root) before = inputs.read(prepared.reviewed_inputs) wait_before = reviewing.goals.goal_wait("owner") @@ -244,29 +258,48 @@ def test_terminal_reply_read_cannot_clear_a_newer_wait(reviewing): assert reviewing.registry.require("owner").goal == goal -@pytest.mark.parametrize("change", ("wait", "turn")) -def test_completed_reply_rechecks_its_wait_and_original_turn(reviewing, change): - reviewing.goals.update_goal("owner", standby(reviewing)) - reply = reviewing.messaging.send("child", "owner", "The actual completed reply").message - witness = reviewing.agents.begin_turn("owner", "original-handling-turn") - wait = reviewing.goals.goal_wait("owner") +def completed_reply(comms): + comms.goals.update_goal("owner", standby(comms)) + reply = comms.messaging.send("child", "owner", "The actual completed reply").message + witness = comms.agents.begin_turn("owner", "original-handling-turn") + return witness, reply, comms.goals.goal_wait("owner") + + +def test_completed_reply_rechecks_its_exact_wait(reviewing): + witness, reply, wait = completed_reply(reviewing) newer = replace(wait, wait_id="newer-wait") + reviewing.bus.bus = AfterReadBus( + reviewing.bus.bus.path, lambda: reviewing.goals.waits.record(newer), + ) + assert not reviewing.goals.consume_reply_wait(witness, reply.reference) + assert reviewing.goals.goal_wait("owner") == newer + + +def test_completed_reply_rechecks_its_original_turn(reviewing): + witness, reply, wait = completed_reply(reviewing) + + def advance_turn(): + reviewing.agents.finish_turn(witness.thread.require_turn_lease()) + reviewing.agents.begin_turn("owner", "a-later-turn") - def replace_state(): - if change == "wait": - reviewing.goals.waits.record(newer) - else: - reviewing.agents.finish_turn(witness.thread.require_turn_lease()) - reviewing.agents.begin_turn("owner", "a-later-turn") + reviewing.bus.bus = AfterReadBus(reviewing.bus.bus.path, advance_turn) + with pytest.raises(StaleFence): + reviewing.goals.consume_reply_wait(witness, reply.reference) + assert reviewing.goals.goal_wait("owner") == wait - reviewing.bus.bus = AfterReadBus(reviewing.bus.bus.path, replace_state) - if change == "wait": - assert not reviewing.goals.consume_reply_wait(witness, reply.reference) - assert reviewing.goals.goal_wait("owner") == newer - else: - with pytest.raises(StaleFence): - reviewing.goals.consume_reply_wait(witness, reply.reference) - assert reviewing.goals.goal_wait("owner") == wait + +def test_capture_keeps_incarnation_across_rename(reviewing): + original = reviewing.registry.require("owner") + reviewing.registry.rename("owner", "renamed-owner") + captured = GoalWaitReview.capture(reviewing.goals, original.incarnation.require_recorded()) + assert captured.owner.name == "renamed-owner" + assert captured.owner.incarnation == original.incarnation + + +def test_probe_observes_actual_exclusive_lock(probed): + with probed.registry.store.locked(): + probed.bus.bus.probe(ReceivedSource.declared_name) + assert probed.bus.bus.checks == [11] @pytest.mark.parametrize("rename", (False, True)) @@ -288,7 +321,9 @@ def test_display_edit_during_standby_review_is_preserved(reviewing, rename): def test_display_edit_during_terminal_review_does_not_strand_wait(reviewing, rename): reviewing.goals.update_goal("owner", standby(reviewing)) fence = reviewing.agents.finish_turn(reviewing.registry.require("child").require_turn_lease()) - captured = GoalWaitReview.capture(reviewing.goals, "owner") + captured = GoalWaitReview.capture( + reviewing.goals, reviewing.registry.require("owner").incarnation.require_recorded(), + ) title = GivenThreadText("Keep this concurrent title") reviewing.registry.register(replace(reviewing.registry.require("owner"), title=title)) if rename: diff --git a/tests/test_owner_restart_handoff.py b/tests/test_owner_restart_handoff.py index b2b2e4f58c..adf2cddd8d 100644 --- a/tests/test_owner_restart_handoff.py +++ b/tests/test_owner_restart_handoff.py @@ -70,8 +70,10 @@ def test_retired_witness_cannot_borrow_live_or_replaced_claim(*, thread_registra {"AGENT_COMMS_AGENT_BIN": "/authored/pi"}) owner = RetiredOwnerLaunch.capture_retired(snapshot, thread, launch) assert owner.require_current(snapshot) is thread - selection = OwnerRestartSelection(owner.owner, launch.process, owner.claim_scope.claimed_generation, - launch.interpreter) + selection = OwnerRestartSelection( + owner.authority.owner, launch.process, owner.authority.claim.claimed_generation, + launch.interpreter, + ) with pytest.raises(RelationViolationError): selection.require_current(snapshot) changed = ( @@ -101,7 +103,6 @@ def test_handoff_codec_preserves_complete_stopped_witness_and_original_launch(*, runtime = RestartEnvironment.inherit(environment) original = OwnerRestartHandoff("/authored", (owner,), runtime, ConfiguredBinary(), InheritedArguments(), LaunchRuntime(Path("/target/bin/python"))) encoded = FieldCodec.encode(original) - assert set(encoded["owners"][0]) == {"owner", "admission", "launch"} decoded = FieldCodec.decode(OwnerRestartHandoff, encoded) assert decoded == original assert decoded.owners[0].require_current(snapshot) is thread diff --git a/tests/test_reservation_rules.py b/tests/test_reservation_rules.py index 036a474f02..80b26cd852 100644 --- a/tests/test_reservation_rules.py +++ b/tests/test_reservation_rules.py @@ -4,10 +4,13 @@ import pytest -from stored_inputs import stored_inputs from agent_comms.field_codec import FieldCodec from agent_comms.input_disposition import InputDispositions +from agent_comms.native_session_files import NativeSessionFiles +from agent_comms.native_session_reopen import NativeSessionIdentity from agent_comms.session_revision import SessionRevision +from agent_comms.thread_identity import RecordedThreadIncarnation +from stored_inputs import stored_inputs def test_session_observation_preserves_source_and_distinguishes_unreadable_proof(tmp_path): @@ -21,22 +24,27 @@ def test_session_observation_preserves_source_and_distinguishes_unreadable_proof session = tmp_path / "session.jsonl" session.write_text("{}\n") - absent = SessionRevision.observe(str(session)).require_available() + files = NativeSessionFiles.for_identity( + NativeSessionIdentity("revision-test", str(session)), + root=tmp_path / "bus", owner=RecordedThreadIncarnation("owner", 1.0), + ) + absent = SessionRevision.observe(files).require_available() assert isinstance(absent.input_proof, MissingInputProofRevision) - assert absent.current(str(session)) + assert absent.current(files) assert absent.native.size == session.stat().st_size - proof = session.with_name(session.name + ".input-proof") + proof = files.input_proof + files.prepare_private_directory(proof.parent) with sqlite3.connect(proof) as db: db.execute("CREATE TABLE original_context (input_id TEXT)") - present = SessionRevision.observe(str(session)).require_available() + present = SessionRevision.observe(files).require_available() assert isinstance(present.input_proof, PresentInputProofRevision) - assert not absent.current(str(session)) + assert not absent.current(files) assert not present.matches(absent) assert FieldCodec.decode(SessionRevision, FieldCodec.encode(present)) == present session.write_text("{}\n{}\n") - appended = SessionRevision.observe(str(session)).require_available() + appended = SessionRevision.observe(files).require_available() assert appended.input_proof == present.input_proof - assert not present.current(str(session)) + assert not present.current(files) encoded = FieldCodec.encode(present) encoded["native"]["size"] = True with pytest.raises(ValueError): @@ -46,13 +54,13 @@ def test_session_observation_preserves_source_and_distinguishes_unreadable_proof # A real stat error must not be called absent or match a reserved source. proof.rename(tmp_path / "original-proof.sqlite3") proof.symlink_to(proof.name) - unavailable = SessionRevision.observe(str(session)) + unavailable = SessionRevision.observe(files) assert not unavailable.matches(present) with pytest.raises(SessionRevisionUnavailable) as error: unavailable.require_available() assert isinstance(error.value.__cause__, OSError) session.unlink() - assert not present.current(str(session)) + assert not present.current(files) def test_plural_originals_preserve_order_and_bind_without_partial_transition(tmp_path): From a43f3f8186362ea21f433b78f48e79bef36a0b4e Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Fri, 23 Oct 2026 03:12:31 -0400 Subject: [PATCH 5/5] Run goal reply locking independently on main with bounded owner queries --- src/agent_comms/bus.py | 4 ++-- src/agent_comms/goal_waits.py | 7 ++++++- tests/test_goal_reply_lock_scope.py | 5 ++--- tests/test_goal_waits.py | 4 +++- 4 files changed, 13 insertions(+), 7 deletions(-) diff --git a/src/agent_comms/bus.py b/src/agent_comms/bus.py index 12b91ceb3a..b13fd22ddd 100644 --- a/src/agent_comms/bus.py +++ b/src/agent_comms/bus.py @@ -388,8 +388,8 @@ def __init__(self, path: Path): def received_after(self, lookup: str, after: int) -> bool: """Whether this recipient has a newer candidate; read one indexed metadata row.""" - with closing(self._received_sequences(lookup, after=after, limit=1)) as positions: - return any(positions) + with self.observing() as db: + return ReceivedSource.exists(db, where="{seq} > ?", parameters=(after,), lookup=lookup) @classmethod def install(cls, path: Path, *, history: Iterable[Message] = ()) -> Self: diff --git a/src/agent_comms/goal_waits.py b/src/agent_comms/goal_waits.py index f0dda22670..255ff31126 100644 --- a/src/agent_comms/goal_waits.py +++ b/src/agent_comms/goal_waits.py @@ -450,7 +450,12 @@ def clear(self, goal_id: str, *, wait_id: str | None = None) -> bool: ).rowcount > 0 def resolve(self, goal: GoalDeclaration) -> GoalWait: - return self.for_goal(goal, self.read()) + with self.snapshot() as db: + for row in StoredGoalWaitRow.select( + db, where="{goal_id} = ?", parameters=(goal.active().reported_id(),), + ): + return row.wait + return NoGoalWait() @staticmethod def for_goal(goal: GoalDeclaration, rows: dict[str, RecordedGoalWait]) -> GoalWait: diff --git a/tests/test_goal_reply_lock_scope.py b/tests/test_goal_reply_lock_scope.py index d46708a1dd..daf0e22269 100644 --- a/tests/test_goal_reply_lock_scope.py +++ b/tests/test_goal_reply_lock_scope.py @@ -40,9 +40,8 @@ def __init__(self, path, registry_path): self.checks = [] def probe(self, statement): - if self.checks or not any(table in statement for table in ( - ReceivedSource.declared_name, Delivered.table.declared_name, - )): + tables = (*ReceivedSource.members_with(ReceivedSource), Delivered.table) + if self.checks or not any(table.declared_name in statement for table in tables): return with self.registry_lock.open("rb") as lock: try: diff --git a/tests/test_goal_waits.py b/tests/test_goal_waits.py index f73dd7fec6..d1b3b5a1f8 100644 --- a/tests/test_goal_waits.py +++ b/tests/test_goal_waits.py @@ -30,5 +30,7 @@ def refuse_full_read(self): guard.setattr(RecordedGoalWait, "__post_init__", decoded_wait) guard.setattr(GoalWaits, "read", refuse_full_read) assert comms.goals.release_waits_after_terminal_turn(fence) == ("owner",) - assert decoded == [goal.id] + # Capturing and validating this same wait may each decode it; unrelated + # rows must never be decoded by either read. + assert set(decoded) == {goal.id} assert len(comms.goals.waits.read()) == 200