diff --git a/src/agent_comms/bus.py b/src/agent_comms/bus.py index 25f0c9b891..b13fd22ddd 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 self.observing() as db: + return ReceivedSource.exists(db, where="{seq} > ?", parameters=(after,), lookup=lookup) + @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 6ae4a7902e..38c3bb83eb 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(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), + 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, ctx.thread) 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..6c8f7ade16 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, @@ -27,6 +27,7 @@ from .goal_waits import ( GoalInputReview, GoalWait, + GoalWaitReview, GoalWaits, RecordedGoalWait, StandbyReview, @@ -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 @@ -51,7 +53,7 @@ GoalWaitTarget, IdleGoalExecution, ) -from .goals import Goal, GoalDeclaration +from .goals import GoalDeclaration from .input_attempt import ( UNOBSERVED_DELIVERY_SCOPE, InputDeliveryScope, @@ -172,23 +174,26 @@ def goal_input_review(self, name: str, goal_id: str, wait_for: Sequence[str]) -> def _goal_input_review( self, thread: Thread, goal_id: str, wait_for: Sequence[str] ) -> GoalInputReview: + captured = GoalWaitReview.capture(self, thread.incarnation.require_recorded()) + thread = captured.owner + 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.") - 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( - goal, thread.incarnation.require_recorded(), targets + incarnation = thread.incarnation.require_recorded() + lookup = incarnation.lookup + after_seq = captured.wait.review_after_seq( + goal, incarnation, targets ) sequences = tuple({ sequence for row in unknown.values() @@ -197,6 +202,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 +240,12 @@ def _goal_input_review( unknown[key] = pending eligible.add(key) return GoalInputReview( - goal_id, + captured.snapshot, + thread, + captured.wait, + captured.turn, + through_seq, targets, - owners, tuple(sorted(unknown.values(), key=lambda row: row.order)), frozenset(eligible), ) @@ -252,106 +261,103 @@ 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: + incarnation = self.registry.require(name).incarnation.require_recorded() + review = GoalWaitReview.capture(self, incarnation) + 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.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." + closed = GoalWaits.closed_wait_group( + owner.name, review.wait.targets, self.waits.read(), snapshot, self.agents, ) - 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), - ) - 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. + 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, ...]: - """Release waits after the latest exact, still-idle child turn. + """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.incarnation.require_recorded()) + except UnregisteredThreadError: + continue + if review.wait == wait: + released.extend(self._release_finished_wait(review, fence)) + return tuple(released) - 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: + 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 () + if any(self.agents.is_working(target.incarnation, review.snapshot) + for target in wait.targets): + 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 () - (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) + 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_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()): - return False - if any(self.agents.is_working(target.incarnation, snapshot) for target in wait.targets): - return False - if wait.has_reply(self.bus.bus, owner.incarnation.require_recorded()): - 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." + 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)), + 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 +401,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.incarnation.require_recorded()) + 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..255ff31126 100644 --- a/src/agent_comms/goal_waits.py +++ b/src/agent_comms/goal_waits.py @@ -3,26 +3,31 @@ 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 contextmanager from dataclasses import dataclass, field 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 +from .errors import UnregisteredThreadError from .field_codec import NullMember from .goal_presentation import GoalExecution, GoalExecutionSource, GoalWaitTarget, StandbyExecution from .goals import Goal, GoalDeclaration from .input_attempt import StoredInput, UnresolvedInput from .locked_store import WireWakeFile +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 .goal_management import Goals from .registry_document import RegistrySnapshot from .threads import Thread @@ -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,15 +254,99 @@ 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 + snapshot: RegistrySnapshot + owner: Thread + wait: GoalWait + turn: TurnLease + + @staticmethod + def capture(goals: Goals, incarnation: RecordedThreadIncarnation) -> GoalWaitReview: + with goals.registry.store.locked(mode=SharedLock()): + snapshot = goals.registry.store.read_snapshot_unlocked() + 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.""" + snapshot = goals.registry.snapshot() + return any( + 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.""" + return ( + actual.same_launch(expected) + 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." + ) + + @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. + if goals.bus.bus.received_after(owner.lookup, through_seq): + continue + yield True + return + + @property + def goal_id(self) -> str: + return self.owner.goal.require().id + + @property + def owners(self) -> frozenset[str]: + return self.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] + 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] = [] @@ -355,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/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/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..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 @@ -875,6 +876,19 @@ 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.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/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()) diff --git a/tests/test_goal_reply_lock_scope.py b/tests/test_goal_reply_lock_scope.py new file mode 100644 index 0000000000..daf0e22269 --- /dev/null +++ b/tests/test_goal_reply_lock_scope.py @@ -0,0 +1,334 @@ +"""Reply reads leave the registry free, and publication keeps its observed state.""" + +import fcntl +import os +from contextlib import contextmanager +from dataclasses import replace +from functools import partial + +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.coordination_errors import StaleFence +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.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 + + +class RegistryProbeBus(Bus): + """A second open file description tries 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): + 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: + 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): + with super().observing() as db: + db.set_trace_callback(self.probe) + try: + yield db + finally: + 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") + 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): + 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] + + +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() + witness = probed.agents.begin_turn("owner", "handling-reply") + 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}",) + + +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", ( + 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, + ) + change(reviewing, prepared) + inputs = InputDispositions(reviewing.root) + 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.read(prepared.reviewed_inputs) == before + assert reviewing.goals.goal_wait("owner") == wait_before + assert reviewing.registry.require("owner").goal == goal_before + + +@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)) + 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 + + +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") + + 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 + + +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)) +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(): + 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, 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: + 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 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 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):