Integrate #1987 #1989 #1923 #1980

src/agent_comms/checkpoint_seals.py

@@ -8,7 +8,7 @@
 from typing import TYPE_CHECKING, Literal
 
 from .declared_family import DeclaredFamily
-from .errors import RelationViolationError
+from .errors import CheckpointNeedsRepairError, RelationViolationError
 from .field_codec import Id128Text, Sha256Text
 from .private_path import FileRevision
 
@@ -148,6 +148,4 @@ def require_rebuilt(self, saved: PrefixWitness, rebuilt: PrefixWitness) -> None:
             raise RelationViolationError("Private bus checkpoint pending suffix differs from intent.")
 
     def require_final(self, saved: PrefixWitness, db_path: Path) -> None:
-        from .private_bus_checkpoint import CheckpointNeedsRepair
-
-        raise CheckpointNeedsRepair("Private bus checkpoint has an unfinished write.")
+        raise CheckpointNeedsRepairError("Private bus checkpoint has an unfinished write.")

src/agent_comms/child_process.py

@@ -1024,15 +1024,34 @@ def decoded(stdout: bytes) -> CheckPrinted:
     def unproven(self, command: str, expected: int) -> tuple[str, ...]: ...
 
     @abstractmethod
-    def distance(self, expected: int) -> int: ...
+    def distance(self, expected: int) -> CheckPrinted: ...
+
+    @abstractmethod
+    def plus(self, amount: int) -> CheckPrinted: ...
+
+    @abstractmethod
+    def require(self, expected: object) -> None: ...
+
+    @abstractmethod
+    def project[T](self, counted: Callable[[int], T], uncounted: Callable[[], T]) -> T: ...
 
 
 @dataclass(frozen=True)
 class CountPrinted(CheckPrinted):
     count: int
 
-    def distance(self, expected: int) -> int:
-        return abs(self.count - expected)
+    def distance(self, expected: int) -> CheckPrinted:
+        return CountPrinted(abs(self.count - expected))
+
+    def plus(self, amount: int) -> CheckPrinted:
+        return CountPrinted(self.count + amount)
+
+    def require(self, expected: object) -> None:
+        if self.count:
+            raise ValueError(f"The order check is {self.count} away from {expected}.")
+
+    def project[T](self, counted: Callable[[int], T], uncounted: Callable[[], T]) -> T:
+        return counted(self.count)
 
     def unproven(self, command: str, expected: int) -> tuple[str, ...]:
         return ((f"Check already prints {self.count} before any work: `{command}` proves nothing; "
@@ -1041,8 +1060,17 @@ def unproven(self, command: str, expected: int) -> tuple[str, ...]:
 
 @dataclass(frozen=True)
 class NoIntegerPrinted(CheckPrinted):
-    def distance(self, expected: int) -> int:
-        raise ValueError(f"The order check printed no integer, so no count of {expected}.")
+    def distance(self, expected: int) -> CheckPrinted:
+        return self
+
+    def plus(self, amount: int) -> CheckPrinted:
+        return self
+
+    def require(self, expected: object) -> None:
+        raise ValueError(f"The order check cannot prove {expected}: it printed no integer.")
+
+    def project[T](self, counted: Callable[[int], T], uncounted: Callable[[], T]) -> T:
+        return uncounted()
 
     def unproven(self, command: str, expected: int) -> tuple[str, ...]:
         return ("the check printed no integer; use grep -c PATTERN FILE or ... | wc -l",)

src/agent_comms/envelope_claim_transitions.py

@@ -257,7 +257,7 @@ def apply(self, previous: ClaimProjection) -> ClaimProjection:
                 raise ClaimTransitionError("Only the exact owner incarnation/generation may release.")
         for resource in self.claims:
             if existing := next_state.get(resource):
-                raise ClaimConflict(existing)
+                existing.require_holder(self.owner, self.incarnation)
         for release in self.releases:
             del next_state[release.resource]
         for resource in self.claims:
@@ -278,6 +278,10 @@ class ClaimOwner:
     def __post_init__(self) -> None:
         FieldCodec.require_fields(self)
 
+    def require_holder(self, owner: str, incarnation: str) -> None:
+        if (self.owner, self.incarnation) != (owner, incarnation):
+            raise ClaimConflict(self)
+
 
 @dataclass(frozen=True, slots=True)
 class ClaimProjection(Mapping[str, ClaimOwner]):

src/agent_comms/errors.py

@@ -42,3 +42,7 @@ def __init__(self, wire_root_id: str, wire_seq: int, message_id: str) -> None:
 
 class HumanClaimBlockedError(RelationViolationError):
     """A prior private reservation cannot be safely retried."""
+
+
+class CheckpointNeedsRepairError(Exception):
+    """A shared source requires exclusive checkpoint repair before it can open."""

src/agent_comms/goals.py

@@ -15,8 +15,8 @@
 from pathlib import Path
 from typing import TYPE_CHECKING, ClassVar, Literal
 
-from .child_process import (BoundedRun, ChildOutcome, ChildResult, ExitedOutcome, SignaledOutcome,
-                            TimedOutOutcome)
+from .child_process import (BoundedRun, CheckPrinted, ChildOutcome, ChildResult, CountPrinted,
+                            ExitedOutcome, SignaledOutcome, TimedOutOutcome)
 from .declared_family import DeclaredFamily
 from .errors import RelationViolationError
 from .field_codec import (
@@ -230,19 +230,17 @@ def require_identity(self, goal_id: str) -> None:
 
 class CheckResult(DeclaredFamily, affix="CheckResult"):
     def require(self, result: ChildResult) -> None:
-        distance = self.distance(result)
-        if distance:
-            raise ValueError(f"The order check is {distance} away from {self}.")
+        self.distance(result).require(self)
 
     @abstractmethod
-    def distance(self, result: ChildResult) -> int:
-        """How far this run is from the result the order requires; zero when it is met."""
+    def distance(self, result: ChildResult) -> CheckPrinted:
+        """The measured distance, or the run's declared missing-count state."""
 
 
 @dataclass(frozen=True)
 class SilentCheckResult(CheckResult):
-    def distance(self, result: ChildResult) -> int:
-        return int(result.outcome != ExitedOutcome(0) or bool(result.stdout))
+    def distance(self, result: ChildResult) -> CheckPrinted:
+        return CountPrinted(int(result.outcome != ExitedOutcome(0) or bool(result.stdout)))
 
 
 @dataclass(frozen=True)
@@ -253,7 +251,7 @@ def __post_init__(self):
         if self.expected < 0:
             raise ValueError("An expected count must be nonnegative.")
 
-    def distance(self, result: ChildResult) -> int:
+    def distance(self, result: ChildResult) -> CheckPrinted:
         return result.outcome.printed_count(result.stdout).distance(self.expected)
 
 

src/agent_comms/messaging.py

@@ -198,6 +198,5 @@ def acknowledge(
         return basis.message_count
 
     def acknowledge_through(self, name: str, sequence: int) -> None:
-        # The bus owns capture under its read lock and the ledger validates that
-        # source. Nesting an exclusive wire hold deadlocks its snapshot reader.
+        # The bus owns capture under its read lock and the ledger validates that source.
         self.bus.mark_delivered_through(self.registry.require(name).name, sequence)

src/agent_comms/private_bus_checkpoint.py

@@ -33,7 +33,7 @@
     ClaimOwner, ClaimProjection, ClaimTransition, ClaimTransitionError,
 )
 from .coordination_tables.publications import PublicationIntents
-from .errors import HumanClaimBlockedError, RelationViolationError
+from .errors import CheckpointNeedsRepairError, HumanClaimBlockedError, RelationViolationError
 from .private_path import FileIdentity, FileRevision
 from .store_files import ExclusiveLock, LockMode, SharedLock
 from .field_codec import FieldCodec, MalformedRecord
@@ -100,7 +100,7 @@ def require_source(self, marker: WireMetadata, info: os.stat_result, stream: Bin
         if self.through_seq > marker.last_seq:
             raise RelationViolationError("Private bus checkpoint exceeds the durable sequence marker.")
         if self.revision.identity != FileIdentity.from_stat(info):
-            raise CheckpointNeedsRepair("Private bus source was relocated.")
+            raise CheckpointNeedsRepairError("Private bus source was relocated.")
 
     def require_append(self, marker: WireMetadata, info: os.stat_result,
                        offset: int, sequence: int) -> None:
@@ -1318,23 +1318,14 @@ def collect(offset, raw, record):
     return expected
 
 
-class CheckpointNeedsRepair(Exception):  # noqa: N818 - names a state
-    """A reader holding the shared bus lock found rows the checkpoint does not
-    cover yet, or an unfinished checkpoint write.
-
-    Only a holder of the exclusive bus lock may repair the checkpoint, so the
-    reader takes the exclusive lock and checks again.
-    """
-
-
 def _check_final_checkpoint_read_only(bus, marker, db, stream, path) -> PrefixWitness:
     """Check a final, up-to-date checkpoint under the shared bus lock; never writes."""
     saved = _saved(db)
     info = os.fstat(stream.fileno())
     saved.require_source(marker, info, stream)
     marker.seal.require_final(saved, path)
     if FileRevision.from_stat(info) != saved.revision:
-        raise CheckpointNeedsRepair("Bus has rows the checkpoint does not cover yet.")
+        raise CheckpointNeedsRepairError("Bus has rows the checkpoint does not cover yet.")
     return saved
 
 
@@ -1369,7 +1360,7 @@ def _verify_open_checkpoint_unlocked(bus, marker, db, stream, path) -> PrefixWit
     info = os.fstat(stream.fileno())
     try:
         saved.require_source(marker, info, stream)
-    except CheckpointNeedsRepair:
+    except CheckpointNeedsRepairError:
         # The copied index rows point at the old file; rebuild them.
         return _rebuild_checkpoint_unlocked(bus, marker, db, path, info, marker.seal)
     recovered = marker.seal.recover(bus, marker, db, path, saved, info)
@@ -1453,7 +1444,7 @@ def opened_private_checkpoint_unlocked(bus: WireLog, marker: WireMetadata, *,
     """Keep only this canonical lock's verified source/index resources open.
 
     Under the shared bus lock other readers run concurrently, so the check is
-    read-only and raises CheckpointNeedsRepair where an exclusive holder would
+    read-only and raises CheckpointNeedsRepairError where an exclusive holder would
     repair the checkpoint.
     """
     path = _path(bus.path)

src/agent_comms/selected_claim.py

@@ -22,14 +22,17 @@
 from .coordination_errors import StaleFence
 from .coordination_tables.participants import OwnerGenerations
 from .envelope_claim_transitions import (
-    ClaimConflict,
     ClaimOwner,
     ExistingFileClaim,
     FileClaimPath,
     WakeClaim,
 )
 from .errors import RelationViolationError
 from .messages import ClaimNoticeKind
+from .message_bus import MessageBus
+from .publisher import Publisher
+from .registry_document import RegistrySnapshot
+from .wire_metadata import WireMetadata
 from .native_input_owner import RegistryOwner
 from .private_registry_guard import _require_no_private_owner_rename
 from .store_files import SharedLock, _store_lock
@@ -125,7 +128,9 @@ def _committed_source(
 @contextmanager
 def _selected_claim_boundary(
     comms: Comms, store: Coordination, claim_scope: WakeClaim, owner_name: str
-):
+) -> Iterator[
+    tuple[MessageBus[Publisher], CheckedWireStoreLock, WireMetadata, RegistrySnapshot, Thread]
+]:
     """One wire/bus/registry boundary for selected claim acquisition."""
     if store.session.path.resolve() != (comms.root / "coordination.sqlite3").resolve():
         raise IdentityConflict("Wake coordinator does not belong to this wire root")
@@ -179,13 +184,10 @@ def publish_selected_resource_claim(
         with bus_lock.current_source(bus.log, metadata) as source:
             existing = source.claim(resource)
             if existing is not None:
-                if (
-                    existing.claim_scope == claim_scope
-                    and existing.incarnation == str(owner.created_at)
-                    and existing.owner == owner.name
-                ):
+                existing.require_holder(owner.name, str(owner.created_at))
+                # Same-operation lookup; holder identity was already checked above.
+                if claim_scope == existing.claim_scope:
                     return existing
-                raise ClaimConflict(existing)
             if source.claim_operation_used(claim_scope.operation_id):
                 raise IdentityConflict("Wake claim operation was already consumed")
         committed = bus.publisher._publish_claim_envelope_unlocked(
@@ -198,15 +200,7 @@ def publish_selected_resource_claim(
         )
         transition = committed.claim_transition
         assert transition is not None and transition.generation is not None
-        return ClaimOwner(
-            resource,
-            transition.owner,
-            transition.incarnation,
-            transition.generation,
-            committed.seq,
-            committed.message_id,
-            claim_scope,
-        )
+        return transition.claimed(resource)
 
 
 @contextmanager