diff --git a/src/agent_comms/checkpoint_seals.py b/src/agent_comms/checkpoint_seals.py index 531f8b2f08..30327b35a3 100644 --- a/src/agent_comms/checkpoint_seals.py +++ b/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.") diff --git a/src/agent_comms/child_process.py b/src/agent_comms/child_process.py index c327025401..f12253f903 100644 --- a/src/agent_comms/child_process.py +++ b/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",) diff --git a/src/agent_comms/envelope_claim_transitions.py b/src/agent_comms/envelope_claim_transitions.py index 610b0749c6..e64fb29ccc 100644 --- a/src/agent_comms/envelope_claim_transitions.py +++ b/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]): diff --git a/src/agent_comms/errors.py b/src/agent_comms/errors.py index fa7e60689a..73eeb92824 100644 --- a/src/agent_comms/errors.py +++ b/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.""" diff --git a/src/agent_comms/goals.py b/src/agent_comms/goals.py index 05055b3114..85653a9dcc 100644 --- a/src/agent_comms/goals.py +++ b/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) diff --git a/src/agent_comms/messaging.py b/src/agent_comms/messaging.py index e202a27731..1c8828c54f 100644 --- a/src/agent_comms/messaging.py +++ b/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) diff --git a/src/agent_comms/private_bus_checkpoint.py b/src/agent_comms/private_bus_checkpoint.py index 80f874cdb7..7687dd4816 100644 --- a/src/agent_comms/private_bus_checkpoint.py +++ b/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,15 +1318,6 @@ 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) @@ -1334,7 +1325,7 @@ def _check_final_checkpoint_read_only(bus, marker, db, stream, path) -> PrefixWi 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) diff --git a/src/agent_comms/selected_claim.py b/src/agent_comms/selected_claim.py index ec0b4ab623..ae7bde4946 100644 --- a/src/agent_comms/selected_claim.py +++ b/src/agent_comms/selected_claim.py @@ -22,7 +22,6 @@ from .coordination_errors import StaleFence from .coordination_tables.participants import OwnerGenerations from .envelope_claim_transitions import ( - ClaimConflict, ClaimOwner, ExistingFileClaim, FileClaimPath, @@ -30,6 +29,10 @@ ) 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 diff --git a/src/agent_comms/store_files.py b/src/agent_comms/store_files.py index 512f4f9abb..d6a415e143 100644 --- a/src/agent_comms/store_files.py +++ b/src/agent_comms/store_files.py @@ -8,11 +8,13 @@ import os import stat import tempfile +import threading import time from abc import ABC, abstractmethod -from collections.abc import AsyncIterator, Iterator, Mapping +from collections.abc import AsyncIterator, Awaitable, Callable, Iterator, Mapping from contextlib import ( AbstractAsyncContextManager, + AbstractContextManager, ExitStack, asynccontextmanager, contextmanager, @@ -23,9 +25,10 @@ from pathlib import Path from typing import TYPE_CHECKING, BinaryIO, ClassVar +from .child_process import Platform from .declared_family import DeclaredFamily -from .errors import RelationViolationError -from .private_path import PrivateFileRole +from .errors import CheckpointNeedsRepairError, RelationViolationError +from .private_path import FileIdentity, PrivateFileRole with suppress(ModuleNotFoundError): # Windows; Win32Platform never asks a mode for its flock. import fcntl @@ -33,9 +36,8 @@ if TYPE_CHECKING: import sqlite3 - from .child_process import Platform from .private_bus_checkpoint import ( - CheckpointNeedsRepair, + CheckedRead, PrefixWitness, WritableCheckpoint, ) @@ -64,8 +66,11 @@ def verify_checkpoint(self, checkpoint: WritableCheckpoint, bus: WireLog, def require_writer(self) -> None: """Require the exclusive holder before rebuilding a replaced source.""" + def require_held(self, held: LockMode) -> None: + pass + @abstractmethod - def after_repair_needed(self, error: CheckpointNeedsRepair) -> LockMode: + def after_repair_needed(self, error: CheckpointNeedsRepairError) -> LockMode: """The mode that may repair what this holder could only name.""" @@ -84,10 +89,9 @@ def verify_checkpoint(self, checkpoint: WritableCheckpoint, bus: WireLog, return checkpoint.check_final(bus, marker, db, stream) def require_writer(self) -> None: - from .private_bus_checkpoint import CheckpointNeedsRepair - raise CheckpointNeedsRepair("Replaced bus file needs its page index rebuilt.") + raise CheckpointNeedsRepairError("Replaced bus file needs its page index rebuilt.") - def after_repair_needed(self, error: CheckpointNeedsRepair) -> LockMode: + def after_repair_needed(self, error: CheckpointNeedsRepairError) -> LockMode: return ExclusiveLock() @@ -108,17 +112,21 @@ def verify_checkpoint(self, checkpoint: WritableCheckpoint, bus: WireLog, def require_writer(self) -> None: pass - def after_repair_needed(self, error: CheckpointNeedsRepair) -> LockMode: + def require_held(self, held: LockMode) -> None: + try: + held.require_writer() + except CheckpointNeedsRepairError as error: + raise RelationViolationError( + "Cannot upgrade an enclosing shared store lock." + ) from error + + def after_repair_needed(self, error: CheckpointNeedsRepairError) -> LockMode: raise error @dataclass(slots=True) class StoreLockContention: - """One operation's borrowed wait resource, spent only at physical locks. - - Source reading, decoding and consumer work cannot spend this resource. - It grants no owner, input, proof or replay permission. - """ + """One operation's contention budget, spent only while acquiring physical locks.""" remaining: float @@ -151,8 +159,6 @@ async def acquire_async(self, descriptor: int, platform: Platform, *, mode: Lock @dataclass class StoreLockRelease: - """Actions owned by one outer physical-lock lifetime.""" - actions: ExitStack = field(default_factory=ExitStack) active: bool = True @@ -166,16 +172,12 @@ def callback(self, operation, /, *args, **kwargs) -> None: def close(self) -> None: if self.active: - # Callbacks may take fresh locks; every inherited context sees this - # scope as ended before the first callback runs. self.active = False self.actions.close() @dataclass(frozen=True) class StoreLock: - """A held store lock's file descriptor.""" - descriptor: int _releases: ClassVar[ContextVar[StoreLockRelease]] = ContextVar( "store_lock_release", default=StoreLockRelease(active=False) @@ -184,7 +186,6 @@ class StoreLock: @classmethod @contextmanager def hold(cls, path: Path, descriptor: int, mode: LockMode): - """Keep a plain store lock; it does not hold a checked bus read.""" yield cls(descriptor) @classmethod @@ -214,12 +215,7 @@ def worker_context(cls) -> Context: @classmethod @contextmanager def releasing(cls): - """Nested locks share the outer lock's post-release work. - - The scope object, not a context-specific token, owns retirement. Marking - it inactive retires every inherited context before callbacks take fresh - locks. A stale inherited value therefore starts an independent scope. - """ + """Share outer retirement; inherited contexts observe that same scope ending.""" current = cls._releases.get() if current.active: yield current @@ -232,15 +228,36 @@ def releasing(cls): scope.close() -class StoreOwner(DeclaredFamily): - """Shared declaration membership for composable store capabilities.""" +class StoreReadBudget: + """The unbounded source-read state.""" + + def check(self, path: Path) -> None: + pass + + +@dataclass(frozen=True) +class BoundedStoreReadBudget(StoreReadBudget): + maximum: int + + def __post_init__(self) -> None: + if type(self.maximum) is not int or self.maximum < 0: + raise ValueError("bus read cap must be a nonnegative integer") + + def check(self, path: Path) -> None: + if path.exists(): + info = path.lstat() + if not stat.S_ISREG(info.st_mode) or info.st_size > self.maximum: + raise RelationViolationError("Bus exceeds bounded read budget.") + +UNBOUNDED_STORE_READ = StoreReadBudget() + + +class StoreOwner(DeclaredFamily): __slots__ = () class StoreLockOwner(StoreOwner, declared_name=None): - """Store transactions use ranks assigned by the complete ordered declaration.""" - __slots__ = () path: Path lock_rank: ClassVar[int] @@ -257,9 +274,9 @@ def declare_lock_order(cls, owners: tuple[type[StoreLockOwner], ...]) -> None: for rank, owner in enumerate(owners): owner.lock_rank = rank - def locked(self, *, blocking: bool = True, max_bus_bytes: int | None = None, + def locked(self, *, blocking: bool = True, read_budget: StoreReadBudget = UNBOUNDED_STORE_READ, mode: LockMode = ExclusiveLock(), contention: StoreLockContention | None = None): - return _store_lock(self.path, blocking=blocking, max_bus_bytes=max_bus_bytes, + return _store_lock(self.path, blocking=blocking, read_budget=read_budget, mode=mode, contention=contention, lock_type=self.lock_type) def locked_async(self) -> AbstractAsyncContextManager[StoreLock]: @@ -273,12 +290,7 @@ def store_lock_path(store_path: Path) -> Path: @contextmanager def _store_lock_file(store_path: Path): - """Own the original lock inode's descriptor before any acquisition. - - The lock file is never written, so it is opened read-only: closing a - descriptor opened for writing is a close-write event that wakes every - owner watching the store directory, once per lock taken. - """ + """Open read-only: lock retirement must not generate a close-write watch event.""" store_path.parent.mkdir(parents=True, exist_ok=True) lock_path = store_lock_path(store_path) with StoreLock.releasing(), os.fdopen( @@ -287,79 +299,168 @@ def _store_lock_file(store_path: Path): yield lock_file -@contextmanager -def _held_store_source(store_path, lock_file, platform, max_bus_bytes, *, mode, lock_type): - """Check durability after locking; release the lock if the check refuses access.""" - try: - if max_bus_bytes is not None: - if type(max_bus_bytes) is not int or max_bus_bytes < 0: - raise ValueError("bus read cap must be a nonnegative integer") - if store_path.exists(): - bus_info = store_path.lstat() - if not stat.S_ISREG(bus_info.st_mode) or bus_info.st_size > max_bus_bytes: - raise RelationViolationError("Bus exceeds bounded read budget.") - with lock_type.hold(store_path, lock_file.fileno(), mode) as lock: +@dataclass(frozen=True) +class SynchronousStoreExecution: + pass + + +@dataclass(frozen=True) +class StoreLockExecution: + pid: int + thread: int + task: asyncio.Task | SynchronousStoreExecution + + +class UnacquiredStoreLock: + @contextmanager + def locked( + self, store_path: Path, read_budget: StoreReadBudget, mode: LockMode, + contention: StoreLockContention, lock_type: type[StoreLock], blocking: bool, + ) -> Iterator[StoreLock]: + platform = Platform.current() + with _store_lock_file(store_path) as lock_file, ExitStack() as resources: + def acquire(mode): + contention.acquire(lock_file.fileno(), platform, mode=mode) + return resources.enter_context(StoreLockAcquisition.holding( + store_path, lock_file, platform, read_budget, mode=mode, lock_type=lock_type)) + try: + lock = acquire(mode) + except CheckpointNeedsRepairError as error: + lock = acquire(mode.after_repair_needed(error)) yield lock - finally: - platform.release_store_lock(lock_file.fileno()) + + @asynccontextmanager + async def locked_async( + self, store_path: Path, read_budget: StoreReadBudget, mode: LockMode, + contention: StoreLockContention, lock_type: type[StoreLock], blocking: bool, + ) -> AsyncIterator[StoreLock]: + platform = Platform.current() + with _store_lock_file(store_path) as lock_file: + await contention.acquire_async(lock_file.fileno(), platform, mode=mode) + with StoreLockAcquisition.holding( + store_path, lock_file, platform, read_budget, mode=mode, lock_type=lock_type, + ) as lock: + yield lock + + async def read_checked[T]( + self, log: WireLog, read: Callable[[CheckedRead], T], mode: LockMode, + worker: Callable[[Callable[[], T]], Awaitable[T]], + ) -> T: + return await log._read_checked_fresh(read, mode, worker) + + +@dataclass +class StoreLockAcquisition: + file: BinaryIO + mode: LockMode + identity: FileIdentity + _held: ClassVar[ContextVar[tuple[tuple[StoreLockExecution, StoreLockAcquisition], ...]]] = ( + ContextVar("store_lock_acquisitions", default=()) + ) + + @staticmethod + def execution() -> StoreLockExecution: + task: asyncio.Task | SynchronousStoreExecution = SynchronousStoreExecution() + match asyncio._get_running_loop(): + case None: + pass + case loop: + match asyncio.current_task(loop): + case asyncio.Task() as running: + task = running + return StoreLockExecution(os.getpid(), threading.get_ident(), task) + + @classmethod + def current( + cls, store_path: Path, mode: LockMode, + ) -> StoreLockAcquisition | UnacquiredStoreLock: + execution = cls.execution() + try: + identity = FileIdentity.from_stat(store_lock_path(store_path).stat()) + except FileNotFoundError: + return UnacquiredStoreLock() + for borrower, acquisition in reversed(cls._held.get()): + if (borrower == execution and acquisition.identity == identity + and not acquisition.file.closed): + mode.require_held(acquisition.mode) + return acquisition + return UnacquiredStoreLock() + + def locked( + self, store_path: Path, read_budget: StoreReadBudget, mode: LockMode, + contention: StoreLockContention, lock_type: type[StoreLock], blocking: bool, + ) -> AbstractContextManager[StoreLock]: + return self.source(store_path, read_budget, lock_type) + + @asynccontextmanager + async def locked_async( + self, store_path: Path, read_budget: StoreReadBudget, mode: LockMode, + contention: StoreLockContention, lock_type: type[StoreLock], blocking: bool, + ) -> AsyncIterator[StoreLock]: + with self.source(store_path, read_budget, lock_type) as lock: + yield lock + + async def read_checked[T]( + self, log: WireLog, read: Callable[[CheckedRead], T], mode: LockMode, + worker: Callable[[Callable[[], T]], Awaitable[T]], + ) -> T: + return await log._read_checked_borrowed(read, self, worker) + + @classmethod + @contextmanager + def holding( + cls, store_path: Path, file: BinaryIO, platform: Platform, read_budget: StoreReadBudget, + *, mode: LockMode, lock_type: type[StoreLock], + ) -> Iterator[StoreLock]: + with ExitStack() as resources: + resources.callback(platform.release_store_lock, file.fileno()) + acquisition = cls(file, mode, FileIdentity.from_stat(os.fstat(file.fileno()))) + yield resources.enter_context(acquisition.source(store_path, read_budget, lock_type)) + + @contextmanager + def source( + self, store_path: Path, read_budget: StoreReadBudget, lock_type: type[StoreLock], + ) -> Iterator[StoreLock]: + """Verify and lend the source, including to an explicitly joined worker.""" + if self.file.closed: + raise RelationViolationError("Store lock acquisition has ended.") + token = self._held.set((*self._held.get(), (self.execution(), self))) + try: + read_budget.check(store_path) + with lock_type.hold(store_path, self.file.fileno(), self.mode) as lock: + yield lock + finally: + self._held.reset(token) -@contextmanager def _store_lock( store_path: Path, *, blocking: bool = True, - max_bus_bytes: int | None = None, + read_budget: StoreReadBudget = UNBOUNDED_STORE_READ, mode: LockMode = ExclusiveLock(), contention: StoreLockContention | None = None, lock_type: type[StoreLock] = StoreLock, -) -> Iterator[StoreLock]: - """Acquire a store lock synchronously; inherited POSIX locks end at last close.""" - from .child_process import Platform - from .private_bus_checkpoint import CheckpointNeedsRepair - - platform = Platform.current() - with _store_lock_file(store_path) as lock_file, ExitStack() as resources: - def acquire(mode): - if contention is None: - platform.acquire_store_lock(lock_file.fileno(), mode=mode, blocking=blocking) - else: - contention.acquire(lock_file.fileno(), platform, mode=mode) - return resources.enter_context(_held_store_source( - store_path, lock_file, platform, max_bus_bytes, mode=mode, lock_type=lock_type)) - - try: - lock = acquire(mode) - except CheckpointNeedsRepair as error: - lock = acquire(mode.after_repair_needed(error)) - yield lock +) -> AbstractContextManager[StoreLock]: + wait = contention or StoreLockContention(math.inf if blocking else 0) + return StoreLockAcquisition.current(store_path, mode).locked( + store_path, read_budget, mode, wait, lock_type, blocking, + ) -@asynccontextmanager -async def _async_store_lock( +def _async_store_lock( store_path: Path, *, blocking: bool = True, - max_bus_bytes: int | None = None, + read_budget: StoreReadBudget = UNBOUNDED_STORE_READ, mode: LockMode = ExclusiveLock(), contention: StoreLockContention | None = None, lock_type: type[StoreLock] = StoreLock, -) -> AsyncIterator[StoreLock]: - """Borrow the same physical resource without blocking its owner event loop. - - Cancellation during acquisition closes only the unacquired descriptor. The - durability check runs only after the lock is acquired; no SQLite - connection or consumer is transferred to another thread. - """ - from .child_process import Platform - - platform = Platform.current() - wait = contention if contention is not None else StoreLockContention(math.inf if blocking else 0) - with _store_lock_file(store_path) as lock_file: - await wait.acquire_async(lock_file.fileno(), platform, mode=mode) - with _held_store_source(store_path, lock_file, platform, max_bus_bytes, - mode=mode, lock_type=lock_type) as lock: - yield lock +) -> AbstractAsyncContextManager[StoreLock]: + wait = contention or StoreLockContention(math.inf if blocking else 0) + return StoreLockAcquisition.current(store_path, mode).locked_async( + store_path, read_budget, mode, wait, lock_type, blocking, + ) def _atomic_write_text( diff --git a/src/agent_comms/todos.py b/src/agent_comms/todos.py index a5edb83dec..6a9d18f50a 100644 --- a/src/agent_comms/todos.py +++ b/src/agent_comms/todos.py @@ -25,7 +25,7 @@ from . import skills from .change_report import FileLine -from .child_process import BoundedRun, ChildResult, ExitedOutcome +from .child_process import BoundedRun, CheckPrinted, ChildResult, ExitedOutcome from .command import Command from .declared_family import DeclaredFamily from .errors import ToolRefusedError @@ -261,7 +261,8 @@ def _clock(at: float) -> str: @dataclass(frozen=True) class DistanceReading: at: float - distance: int + distance: CheckPrinted + command: str = "" @dataclass(frozen=True) @@ -272,15 +273,25 @@ class DistanceReadings: @property def trend(self) -> DistanceTrend: - """Velocity across the two readings: falling projects the finish, else it is flat.""" + """A missing count diagnoses the check; only measured counts project a finish.""" + if not self.readings: + return UnreadTrend() + return self.readings[-1].distance.project( + self.counted_trend, lambda: UncountedTrend(self.readings[-1].command)) + + def counted_trend(self, final: int) -> DistanceTrend: last = self.readings[-2:] - if len(last) < 2 or not last[1].distance: + if len(last) < 2 or not final: return UnreadTrend() - first, final = last - if final.distance >= first.distance: - return FlatTrend(final.distance, final.at) - rate = (first.distance - final.distance) / max(final.at - first.at, 1.0) - return MovingTrend(final.distance, final.at + final.distance / rate) + first, latest = last + + def compared(previous: int) -> DistanceTrend: + if final >= previous: + return FlatTrend(final, latest.at) + rate = (previous - final) / max(latest.at - first.at, 1.0) + return MovingTrend(final, latest.at + final / rate) + + return first.distance.project(compared, UnreadTrend) class DistanceTrend(DeclaredFamily, affix="Trend"): @@ -313,6 +324,15 @@ def projection(self) -> str: return f"; distance {self.distance}, not falling" +@dataclass(frozen=True) +class UncountedTrend(DistanceTrend): + command: str + + @property + def projection(self) -> str: + return f"; check prints no integer: {self.command}" + + @dataclass(frozen=True) class UnreadTrend(DistanceTrend): """Fewer than two readings, or the last one at done: nothing to project from.""" @@ -369,7 +389,7 @@ def readings(self, source: ProgressSource) -> DistanceReadings: "--not", base).splitlines()))] points = [(base, source.since), *commits][-2:] return DistanceReadings( - tuple(DistanceReading(at, source.distance(self, base, sha)) for sha, at in points)) + tuple(source.reading(self, base, sha, at) for sha, at in points)) @dataclass(frozen=True) @@ -400,12 +420,13 @@ def commits(self, *excluded: str) -> Iterator[tuple[str, float]]: def readings(self) -> DistanceReadings: return self.checkout.readings(self) - def distance(self, checkout: PresentCheckout, base: str, sha: str) -> int: - """Run the order's check and count its untouched edits at a present revision.""" + def reading(self, checkout: PresentCheckout, base: str, sha: str, at: float) -> DistanceReading: + """Keep the check command with its measured or missing distance.""" check = self.todo.order.check touched = checkout.git("diff", "--name-only", base, sha).split() run = check.at(sha, str(checkout.path)) - return check.result.distance(run) + self.todo.order.untouched(touched) + return DistanceReading(at, check.result.distance(run).plus( + self.todo.order.untouched(touched)), check.command) @dataclass(frozen=True) @@ -2082,8 +2103,13 @@ def sent_in(self, message_id: str) -> bool: return bool(self.messages) and self.messages[-1].message_id == message_id def require_replacement(self, message_id: str) -> None: + if not self.messages: + raise ValueError(f"{self.holder} holds {self.id}, but no order message was recorded.") if not self.sent_in(message_id): - raise ValueError(f"{self.holder} already holds {self.id}; pass --replaces with its current message.") + raise ValueError( + f"{self.holder} already holds {self.id}; " + f"pass --replaces {self.messages[-1].message_id}" + ) @classmethod def require_order_body(cls, body: str) -> None: diff --git a/src/agent_comms/wire_log.py b/src/agent_comms/wire_log.py index fba44b60c0..4aaa135198 100644 --- a/src/agent_comms/wire_log.py +++ b/src/agent_comms/wire_log.py @@ -10,8 +10,8 @@ import stat import time from abc import ABC, abstractmethod -from collections.abc import Callable, Iterable, Iterator, Mapping -from contextlib import ExitStack, contextmanager +from collections.abc import Awaitable, Callable, Iterable, Iterator, Mapping +from contextlib import AbstractContextManager, ExitStack, contextmanager from dataclasses import dataclass, field, replace from functools import partial from io import DEFAULT_BUFFER_SIZE @@ -23,6 +23,7 @@ ) from .audience_manifest import MAX_WIRE_SEQ +from .child_process import Platform from .bus_publication import ( CommittedDelivery, has_private_wire_fields, @@ -35,6 +36,7 @@ ClaimProjection, ) from .errors import ( + CheckpointNeedsRepairError, RelationViolationError, StoredHistoryUnreadable, ) @@ -51,11 +53,12 @@ StoreLock, StoreLockOwner, StoreLockContention, + StoreLockAcquisition, + UNBOUNDED_STORE_READ, _atomic_write_text, _iter_jsonl_stream, _store_lock, _store_lock_file, - _held_store_source, ) from .wire_metadata import WireMetadata from .wire_record import WireRecord, WireScan @@ -581,42 +584,56 @@ def checked_read(self, *, blocking: bool = True, contention: StoreLockContention yield source async def read_checked_async(self, read: Callable): - """Acquire cancellably, then consume the whole source in its worker. - - Return detached observations only. Verification and the callback share - one resource thread; no SQLite connection crosses to the event loop. - The worker releases its locks before returning its result; result - delivery must not keep a POSIX lock until the event loop resumes. - Cancellation joins acquired work before the original descriptor closes. - """ - from .private_bus_checkpoint import CheckpointNeedsRepair - + """Consume in the joined source worker; cancellation waits for acquired work.""" # Readers share the bus lock so idle workers don't queue behind each # other; only a checkpoint repair needs the exclusive lock. mode = SharedLock() try: return await self._read_checked_under_lock(read, mode) - except CheckpointNeedsRepair as error: + except CheckpointNeedsRepairError as error: return await self._read_checked_under_lock(read, mode.after_repair_needed(error)) - async def _read_checked_under_lock(self, read: Callable, mode: LockMode): - from .child_process import Platform + async def _read_checked_under_lock[T]( + self, read: Callable[[CheckedRead], T], mode: LockMode, + ) -> T: from .coordinator import Coordination + return await StoreLockAcquisition.current(self.path, mode).read_checked( + self, read, mode, Coordination.run_worker, + ) + + def _consume_checked[T]( + self, read: Callable[[CheckedRead], T], acquisition: AbstractContextManager, + ) -> T: + with acquisition as lock, self._checked_source(lock) as source: + return read(source) + + async def _read_checked_borrowed[T]( + self, read: Callable[[CheckedRead], T], held: StoreLockAcquisition, + worker: Callable[[Callable[[], T]], Awaitable[T]], + ) -> T: + return await worker(partial( + self._consume_checked, read, held.source(self.path, UNBOUNDED_STORE_READ, self.lock_type) + )) + + async def _read_checked_fresh[T]( + self, read: Callable[[CheckedRead], T], mode: LockMode, + worker: Callable[[Callable[[], T]], Awaitable[T]], + ) -> T: platform = Platform.current() with _store_lock_file(self.path) as lock_file: await StoreLockContention(math.inf).acquire_async( lock_file.fileno(), platform, mode=mode, ) - return await Coordination.run_worker( - partial(self._read_checked, lock_file, platform, read, mode), - ) - def _read_checked(self, lock_file, platform, read, mode): - with lock_file, _held_store_source(self.path, lock_file, platform, None, - mode=mode, lock_type=self.lock_type) as lock: - with self._checked_source(lock) as source: - return read(source) + def read_acquired() -> T: + with lock_file: + return self._consume_checked(read, StoreLockAcquisition.holding( + self.path, lock_file, platform, UNBOUNDED_STORE_READ, + mode=mode, lock_type=self.lock_type, + )) + + return await worker(read_acquired) @contextmanager def _checked_source(self, lock: CheckedWireStoreLock): diff --git a/tests/test_commander_problem_orders.py b/tests/test_commander_problem_orders.py index 8fd8cd8a24..2d7c880cdc 100644 --- a/tests/test_commander_problem_orders.py +++ b/tests/test_commander_problem_orders.py @@ -477,10 +477,23 @@ def test_commander_script_records_every_order_step_in_core(commander, commander_ 'worker', ('Set count.txt to 1.',), ('no second store', 'no cache')) numbered = body.read_text().replace('Problem: NEW', f'Problem: {created.id}') + # The refusal itself supplies the exact replacement argument; no DM lookup. + with pytest.raises(ValueError) as refusal: + script.send('worker', due, numbered) + expected = f"worker already holds P1; pass --replaces {created.messages[-1].message_id}" + assert str(refusal.value) == expected + replaces = str(refusal.value).split('--replaces ', 1)[1] + print(f'Refused order: {refusal.value}') # Without --clean this replaces the prior order and counts a send-back. - script.send('worker', due, numbered, replaces=created.messages[-1].message_id) + script.send('worker', due, numbered, replaces=replaces) sent_back = comms.goals.todos.get('P1') assert sent_back.order.id != created.order.id and script.struck(sent_back, 'worker') == 1 + print(f'Accepted replacement: {sent_back.order.id}') + # A historical send cannot replace the latest order. + with pytest.raises(ValueError, match=sent_back.messages[-1].message_id): + script.send('worker', due, numbered, replaces=replaces) + with pytest.raises(ValueError, match='no order message was recorded'): + replace(sent_back, messages=()).require_replacement(replaces) (link,) = sent_back.messages[-1:] sent = next(m.body for m in comms.views.history_page(CommandTarget.named('worker'), viewer='lead').messages if m.message_id == link.message_id) assert 'Set count.txt to 1.' in sent and 'Start now and push as you go.' in sent @@ -518,6 +531,40 @@ def problem(script, body): script.problem('The counter file holds zero where the fleet needs one', 'worker', '23:59', body, f'{__file__}:1') +def test_installed_order_entrypoint_refuses_then_accepts_printed_replacement( + commander, commander_script): + from agent_comms.rule_reviews import ( + AnsweredRuleReviewEntry, DrawnRuleReviewEntry, RuleReviews, + ) + + comms, _ = commander + script, repo = commander_script + (repo / 'count.txt').write_text('0\n') + body = Path(order_body(repo)) + problem(script, str(body)) + body.write_text(body.read_text().replace('Problem: NEW', 'Problem: P1')) + reviews = RuleReviews(comms.root) + reviews.write('lead', DrawnRuleReviewEntry(('Keep the current order message with its Todo.',), 0)) + reviews.write('lead', AnsweredRuleReviewEntry( + 1, 'P1 keeps the current message in its Todo and prints it when refusing replacement.')) + command = [str(Path(sys.executable).with_name('agent-comms-commander')), + 'order', 'send', 'worker', '23:59', str(body)] + env = {**os.environ, 'PYTHONPATH': str(Path(__file__).parents[1] / 'src'), + 'AGENT_COMMS_ROOT': str(comms.root)} + refused = subprocess.run(command, env=env, capture_output=True, text=True, timeout=30) + assert refused.returncode != 0 + current = comms.goals.todos.get('P1') + expected = f'worker already holds P1; pass --replaces {current.messages[-1].message_id}' + assert expected in refused.stderr + replacement = re.search(r'pass --replaces (\w+)', refused.stderr).group(1) + print(expected) + accepted = subprocess.run([*command, '--replaces', replacement], env=env, + capture_output=True, text=True, timeout=30) + assert accepted.returncode == 0, accepted.stderr + assert comms.goals.todos.get('P1').order.id != current.order.id + print(f'Accepted --replaces {replacement}: {accepted.stdout.strip()}') + + def test_resend_after_strike_counts_one_miss(commander, commander_script, monkeypatch): """A miss struck with `order strike` and then sent back is one strike; the next send-back strikes again.""" comms, _ = commander @@ -659,7 +706,7 @@ def test_commander_refuses_a_check_that_prints_no_count(commander_script): @pytest.mark.parametrize('printed, refused', [ - ("printf 'no count here'", 'The order check printed no integer, so no count of 1.'), + ("printf 'no count here'", 'The order check cannot prove CountCheckResult(expected=1): it printed no integer.'), ('echo 2', 'The order check is 1 away from CountCheckResult(expected=1).'), ('echo found 1', None), ]) diff --git a/tests/test_envelope_claim_transitions.py b/tests/test_envelope_claim_transitions.py index 435fbf179c..6927c57214 100644 --- a/tests/test_envelope_claim_transitions.py +++ b/tests/test_envelope_claim_transitions.py @@ -95,6 +95,20 @@ def test_two_serialized_conflicting_entrants_have_one_winner_and_one_synchronous first[a] = None +def test_same_holder_moves_scope_but_another_holder_cannot(tmp_path): + _, resource, _ = resources(tmp_path) + held = transition(1, claims=(resource,), generation=G1).apply(ClaimProjection()) + moved = transition(2, claims=(resource,), generation=G2).apply(held) + assert held[resource].generation == G1 + assert moved[resource].generation == G2 + for changed in ({"owner": "other"}, {"incarnation": "new-birth"}): + with pytest.raises(ClaimConflict): + transition(3, claims=(resource,), generation=G1, **changed).apply(moved) + with pytest.raises(ClaimTransitionError, match="exact owner"): + transition(3, releases=(ClaimRelease(resource, G1),)).apply(moved) + assert dict(transition(3, releases=(ClaimRelease(resource, G2),)).apply(moved)) == {} + + def test_full_set_claim_and_release_are_all_or_nothing(tmp_path): _, a, b = resources(tmp_path) first = transition(1, claims=(b,), generation=G1).apply(ClaimProjection()) diff --git a/tests/test_import_observation.py b/tests/test_import_observation.py new file mode 100644 index 0000000000..1118a82f9e --- /dev/null +++ b/tests/test_import_observation.py @@ -0,0 +1,24 @@ +"""An actual importer and sidebar reader must finish, not lock each other out.""" +import os +from pathlib import Path +import signal +import subprocess +import sys + + +def test_import_and_sidebar_observation_complete_on_a_real_private_bus(tmp_path): + repository = Path(__file__).resolve().parents[1] + process = subprocess.Popen( + [sys.executable, '-X', 'faulthandler', str(repository / 'tools/reproduce_thread_list.py'), + str(tmp_path / 'wire'), '--seconds', '3', '--imports'], + cwd=repository, env={**os.environ, 'PYTHONPATH': str(repository / 'src')}, + stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, start_new_session=True, + ) + try: + output, _ = process.communicate(timeout=25) + print(output) + assert process.returncode == 0, output + finally: + if process.poll() is None: + os.killpg(process.pid, signal.SIGKILL) + process.wait() diff --git a/tests/test_order_progress.py b/tests/test_order_progress.py index 7d8041a799..b676797076 100644 --- a/tests/test_order_progress.py +++ b/tests/test_order_progress.py @@ -66,6 +66,34 @@ def test_no_commit_since_the_order_is_explicit(checkout): assert progress.line == 'no commit since the order' +def test_uncounted_check_keeps_other_overdue_orders_visible(commander, commander_script, checkout, monkeypatch): + from agent_comms.child_process import CountPrinted, NoIntegerPrinted + from agent_comms.todos import DistanceReading, UncountedTrend + + script, _ = commander_script + todo, unit = checkout + comms, _ = commander + monkeypatch.setattr(script.COMMS.registry, 'storage_rule', StorageRule(checkout_base=unit.parent)) + monkeypatch.setattr(PullRequests, 'read', lambda *args: ()) + broken = replace(todo.order, check=replace(todo.order.check, command="printf 'no count here'")) + with comms.goals.todos.writing() as db: + replace(todo, order=broken).upsert(db) + create(commander, 'P2', deadline=1) + commit(unit) + + progress = read((comms.goals.todos.get('P1'), unit)) + assert progress.line.endswith("; check prints no integer: printf 'no count here'") + lines = deadline_lines(script) + alarm = next(line for line in lines if line.startswith('OVERDUE:')) + assert "order order-P1" in alarm and "order order-P2" in alarm + assert "check prints no integer: printf 'no count here'" in alarm + assert 'distance 0' not in alarm and 'projected done' not in alarm + missing = NoIntegerPrinted() + assert missing.distance(1) is missing and missing.plus(3) is missing + assert DistanceReadings((DistanceReading(1, missing, 'printf missing'),)).trend == UncountedTrend('printf missing') + assert DistanceReadings((DistanceReading(1, missing), DistanceReading(2, CountPrinted(2)))).trend.projection == '' + + def test_an_unpushed_commit_names_its_sha_and_time(checkout): sha = commit(checkout[1]) progress = read(checkout) diff --git a/tests/test_private_bus_checkpoint.py b/tests/test_private_bus_checkpoint.py index 7f92296aab..280f55dd27 100644 --- a/tests/test_private_bus_checkpoint.py +++ b/tests/test_private_bus_checkpoint.py @@ -18,13 +18,15 @@ from agent_comms.child_process import ProcessIdentity from agent_comms.comms import Comms from agent_comms.envelope_claim_transitions import ExistingFileClaim -from agent_comms.errors import RelationViolationError +from agent_comms.errors import CheckpointNeedsRepairError, RelationViolationError from agent_comms.messages import Message, HandoffKind, InfoKind from agent_comms.bus_source_page import AddressedPage from agent_comms.private_bus_checkpoint import ( install_private_bus_checkpoint, ) -from agent_comms.store_files import ExclusiveLock, SharedLock, _store_lock +from agent_comms.store_files import ( + BoundedStoreReadBudget, ExclusiveLock, SharedLock, UNBOUNDED_STORE_READ, _store_lock, +) from agent_comms.threads import Thread @@ -470,7 +472,7 @@ def test_over_1000_initials_and_8mib_complete_lookup(tmp_path: Path, monkeypatch # source API supplies a complete-prefix+bounded-page alternative. with ( pytest.raises(RelationViolationError), - comms.bus.log.locked(max_bus_bytes=8 * 1024 * 1024), + comms.bus.log.locked(read_budget=BoundedStoreReadBudget(8 * 1024 * 1024)), ): pass with monkeypatch.context() as patch: @@ -523,8 +525,7 @@ def test_reopened_root_preserves_checked_source_and_supports_append( @pytest.mark.parametrize("copy_root", [False, True]) def test_copied_root_rebuilds_its_index(tmp_path: Path, copy_root: bool) -> None: from agent_comms.child_process import Platform - from agent_comms.private_bus_checkpoint import CheckpointNeedsRepair - from agent_comms.store_files import _store_lock_file + from agent_comms.store_files import StoreLockAcquisition, _store_lock_file original, root_id = _root(tmp_path) first = publish_initial(original, "sender", "Alice", "copied original") @@ -543,8 +544,10 @@ def test_copied_root_rebuilds_its_index(tmp_path: Path, copy_root: bool) -> None platform = Platform.current() with _store_lock_file(log.path) as lock_file: platform.acquire_store_lock(lock_file.fileno(), mode=SharedLock(), blocking=True) - with pytest.raises(CheckpointNeedsRepair, match="relocated"): - log._read_checked(lock_file, platform, lambda source: None, SharedLock()) + with pytest.raises(CheckpointNeedsRepairError, match="relocated"): + with StoreLockAcquisition.holding(log.path, lock_file, platform, UNBOUNDED_STORE_READ, + mode=SharedLock(), lock_type=log.lock_type): + pass # The real reader upgrades its lock and rebuilds the copied SQLite index. request = AddressedPage.capture(lookup=RecordedThreadIncarnation("fixture", 17002.0).lookup) witness, messages, _ = asyncio.run(log.read_checked_async( @@ -651,7 +654,7 @@ def test_shared_reader_hands_lagged_suffix_to_exclusive_repair(tmp_path: Path, m import agent_comms.private_bus_checkpoint as checkpoint from agent_comms.child_process import Platform - from agent_comms.store_files import _store_lock_file + from agent_comms.store_files import StoreLockAcquisition, _store_lock_file comms, _ = _root(tmp_path) publish_initial(comms, "sender", "Alice", "first") @@ -668,8 +671,10 @@ def test_shared_reader_hands_lagged_suffix_to_exclusive_repair(tmp_path: Path, m # A shared-lock reader never writes the checkpoint; it names the repair. with _store_lock_file(log.path) as lock_file: platform.acquire_store_lock(lock_file.fileno(), mode=SharedLock(), blocking=True) - with pytest.raises(checkpoint.CheckpointNeedsRepair): - log._read_checked(lock_file, platform, lambda source: None, SharedLock()) + with pytest.raises(CheckpointNeedsRepairError): + with StoreLockAcquisition.holding(log.path, lock_file, platform, UNBOUNDED_STORE_READ, + mode=SharedLock(), lock_type=log.lock_type): + pass # The async reader then takes the exclusive lock, which repairs and reads. through = asyncio.run(log.read_checked_async(lambda source: source.witness.through_seq)) assert through == 2 @@ -679,11 +684,9 @@ def test_shared_reader_hands_lagged_suffix_to_exclusive_repair(tmp_path: Path, m @pytest.mark.parametrize(("mode", "escalated"), [(SharedLock(), ExclusiveLock), (ExclusiveLock(), None)]) def test_only_a_shared_reader_retries_a_needed_repair_exclusively(mode, escalated) -> None: - from agent_comms.private_bus_checkpoint import CheckpointNeedsRepair - - error = CheckpointNeedsRepair("lagged suffix") + error = CheckpointNeedsRepairError("lagged suffix") if escalated is None: - with pytest.raises(CheckpointNeedsRepair) as raised: + with pytest.raises(CheckpointNeedsRepairError) as raised: mode.after_repair_needed(error) assert raised.value is error else: diff --git a/tests/test_store_lock_order.py b/tests/test_store_lock_order.py index 77b60c319e..32ee92869d 100644 --- a/tests/test_store_lock_order.py +++ b/tests/test_store_lock_order.py @@ -12,7 +12,8 @@ from agent_comms.historical_views import DMDisplayHistory from agent_comms.message_bus import MessageBus from agent_comms.store_files import ( - ExclusiveLock, SharedLock, StoreLock, StoreLockOwner, StoreOwner, store_lock_path, + ExclusiveLock, SharedLock, StoreLock, StoreLockContention, StoreLockOwner, + StoreOwner, store_lock_path, ) from agent_comms.threads import Thread from agent_comms.wire_log import WireLog @@ -67,7 +68,7 @@ def profile(frame, event, arg): ranks[str(store_lock_path(owner.path))] = owner.lock_rank return operation = frame.f_code.co_name - if operation not in {"acquire_store_lock", "_store_lock_file"}: + if operation != "_store_lock_file" and frame.f_code is not StoreLockContention.acquire.__code__: return actor = threading.current_thread().name active = held.setdefault(actor, {}) diff --git a/tests/test_store_lock_reentry.py b/tests/test_store_lock_reentry.py new file mode 100644 index 0000000000..57a1f40431 --- /dev/null +++ b/tests/test_store_lock_reentry.py @@ -0,0 +1,124 @@ +"""Physical locking, not fake store objects: nesting must keep outsiders out.""" +import asyncio +from contextvars import copy_context +import fcntl +from pathlib import Path +import subprocess +import sys + +import pytest + +from agent_comms.comms import wire +from agent_comms.errors import RelationViolationError +from agent_comms.store_files import ( + BoundedStoreReadBudget, SharedLock, StoreLockContention, + _async_store_lock, _store_lock, store_lock_path, +) + + +def probe(path, *, shared=True): + """A fresh interpreter cannot inherit our process-local acquisition context.""" + result = subprocess.run([ + sys.executable, '-c', + 'import fcntl,sys\n' + 'with open(sys.argv[1], "rb") as stream:\n' + ' try: fcntl.flock(stream, int(sys.argv[2]) | fcntl.LOCK_NB)\n' + ' except BlockingIOError: sys.exit(73)\n', + str(store_lock_path(path)), str(fcntl.LOCK_SH if shared else fcntl.LOCK_EX), + ], capture_output=True, text=True, timeout=5) + assert result.returncode in (0, 73), result.stderr + return result.returncode == 0 + + +def test_nested_reader_reuses_descriptor_without_downgrading_or_releasing(tmp_path): + path = tmp_path / 'wire' + with _store_lock(path) as outer: + with pytest.raises(ValueError, match='consumer failed'): + with _store_lock(path, mode=SharedLock(), contention=StoreLockContention(0)) as inner: + assert inner.descriptor == outer.descriptor + links = [] + for descriptor in Path('/proc/self/fd').iterdir(): + try: + links.append(descriptor.readlink()) + except FileNotFoundError: + pass + assert links.count(store_lock_path(path)) == 1 + assert not probe(path) + raise ValueError('consumer failed') + assert not probe(path) + assert probe(path, shared=False) + + +def test_shared_nesting_never_silently_upgrades(tmp_path): + path = tmp_path / 'wire' + with _store_lock(path, mode=SharedLock()) as outer: + with _store_lock(path, mode=SharedLock()) as inner: + assert inner.descriptor == outer.descriptor + assert probe(path) + assert not probe(path, shared=False) + with pytest.raises(RelationViolationError, match='Cannot upgrade'): + with _store_lock(path): + pytest.fail('upgraded the enclosing read transaction') + assert probe(path, shared=False) + + +def test_ended_context_does_not_reuse_its_closed_descriptor(tmp_path): + path = tmp_path / 'wire' + with _store_lock(path): + retained = copy_context() + def capture_again(): + with _store_lock(path): + assert not probe(path) + retained.run(capture_again) + assert probe(path, shared=False) + + +def test_nested_source_still_checks_each_borrows_budget(tmp_path): + path = tmp_path / 'wire' + path.write_bytes(b'not within the nested budget') + with _store_lock(path): + with pytest.raises(RelationViolationError, match='bounded read budget'): + with _store_lock(path, mode=SharedLock(), read_budget=BoundedStoreReadBudget(1)): + pytest.fail('nested source skipped its read budget') + assert not probe(path) + assert probe(path, shared=False) + + +def test_async_and_sync_nest_but_independent_tasks_and_threads_do_not(tmp_path): + path = tmp_path / 'wire' + async def independent(): + with pytest.raises(BlockingIOError): + async with _async_store_lock(path, mode=SharedLock(), blocking=False): + pytest.fail('another task borrowed an exclusive transaction') + def thread_reader(): + with pytest.raises(BlockingIOError): + with _store_lock(path, mode=SharedLock(), blocking=False): + pytest.fail('another thread borrowed an exclusive transaction') + async def run(): + async with _async_store_lock(path) as outer: + async with _async_store_lock(path, mode=SharedLock()) as inner: + assert inner.descriptor == outer.descriptor + with _store_lock(path, mode=SharedLock()) as synchronous: + assert synchronous.descriptor == outer.descriptor + assert not probe(path) + await asyncio.create_task(independent()) + await asyncio.to_thread(thread_reader) + assert not probe(path) + assert probe(path, shared=False) + asyncio.run(run()) + + +def test_checked_async_worker_explicitly_borrows_the_original_acquisition(tmp_path): + comms = wire(tmp_path) + comms.messaging.initialize_private_initial_protocol() + async def run(): + with comms.bus.log.locked() as outer: + def read(source): + # The joined worker may itself nest a normal source acquisition. + with comms.bus.log.locked(mode=SharedLock()) as inner: + assert inner.descriptor == outer.descriptor + return source.witness.through_seq + assert await comms.bus.log.read_checked_async(read) == 0 + assert not probe(comms.bus.log.path) + assert probe(comms.bus.log.path, shared=False) + asyncio.run(run()) diff --git a/tools/debt_ratchet.py b/tools/debt_ratchet.py index eb473e33e4..927126dbd0 100644 --- a/tools/debt_ratchet.py +++ b/tools/debt_ratchet.py @@ -23,9 +23,18 @@ from dataclasses import dataclass from functools import cached_property from pathlib import Path -from refactor_audit.handler_declarations import BuiltinHandlerDeclarations -from refactor_audit.measures import FamilyFlattened as AuditFamilyFlattened -from refactor_audit.measures import ForeignAbsenceProbe as AuditForeignAbsenceProbe + +sys.path.insert(0, str( + Path(__file__).resolve().parents[1] / subprocess.check_output( + ["git", "config", "--file", str(Path(__file__).resolve().parents[1] / ".gitmodules"), + "--get", "submodule.src/agent_comms/skills/refactor-audit.path"], text=True, + ).strip() / "scripts" +)) + +import audit +from audit.handler_declarations import BuiltinHandlerDeclarations +from audit.measures import FamilyFlattened as AuditFamilyFlattened +from audit.measures import ForeignAbsenceProbe as AuditForeignAbsenceProbe from agent_comms.core_stamp import WheelStamp from agent_comms.declared_family import DeclaredFamily @@ -34,6 +43,8 @@ from agent_comms.field_codec import FieldCodec, boundary_decoder, projected from agent_comms.mro_dispatch import MroDispatch, handles +AUDIT_SCRIPTS = Path(audit.__path__[0]).parent + class Measure(DeclaredFamily, affix="Measure"): @staticmethod @@ -228,7 +239,7 @@ def environment(root: str) -> dict[str, str]: Tools import the same revision's ``src`` (``upgrade_bus.py`` puts it first), never the installed one.""" paths = dict.fromkeys((Path(root).parts[0], "src")) - return {**os.environ, "MYPYPATH": os.pathsep.join(paths)} + return {**os.environ, "MYPYPATH": os.pathsep.join((*paths, str(AUDIT_SCRIPTS)))} class SourceLines(Measure): @@ -1448,7 +1459,9 @@ def declares_version(source: bytes) -> bool: @classmethod def formats(cls, repo: Path, ref: str, root: str, changed: set[str]) -> dict[str, StoredForm]: """Each versioned record in the changed modules, measured by that revision's own code.""" - package = Path(root).parent + package = Path(root) + while git(repo, "ls-tree", "--name-only", ref, "--", str(package / "__init__.py")): + package = package.parent modules = [".".join(Path(path).relative_to(package).with_suffix("").parts) for path in sorted(changed & python_paths(repo, ref, root)) if cls.declares_version(git(repo, "show", f"{ref}:{path}"))] @@ -1458,7 +1471,10 @@ def formats(cls, repo: Path, ref: str, root: str, changed: set[str]) -> dict[str source = git(repo, "archive", ref, "--", str(package)) subprocess.run(["tar", "-x", "-C", scratch], input=source, check=True, timeout=60) # Extracted source is stamped with its commit the way a build stamps it. - WheelStamp(revision(repo, ref)).write(Path(scratch) / root / "package_stamp.py") + for module in {module.split(".")[0] for module in modules}: + WheelStamp(revision(repo, ref)).write( + Path(scratch) / package / module / "package_stamp.py" + ) result = subprocess.run([sys.executable, "-c", cls.probe, str(Path(scratch) / package), str(Path(__file__).resolve().parent), *modules], cwd=scratch, env=cls.environment(root), diff --git a/tools/reproduce_thread_list.py b/tools/reproduce_thread_list.py new file mode 100644 index 0000000000..eb013925d3 --- /dev/null +++ b/tools/reproduce_thread_list.py @@ -0,0 +1,114 @@ +from __future__ import annotations + +import argparse +import faulthandler +import json +import multiprocessing as mp +import time +from dataclasses import replace +from multiprocessing.queues import Queue +from multiprocessing.synchronize import Event +from pathlib import Path + +from agent_comms.cli_bridges.claude import ClaudeTranscript +from agent_comms.comms import Comms +from agent_comms.storage_rules import StorageRoots +from agent_comms.thread_status import StoppedThreadStatus +from agent_comms.threads import Thread +from agent_comms.ui_model.observation import ObserveSidebar, SidebarObserved, SidebarUnreadable + + +def declaration(root: Path, name: str, created_at: float = 1) -> Thread: + return Thread( + name, + frozenset({"proof"}), + str(root), + created_at=created_at, + storage_roots=StorageRoots(root / "scratch" / name), + ) + + +def writer(root: Path, stop: Event, result: Queue[int], imports: bool) -> None: + comms = Comms(root) + count = 0 + while not stop.is_set(): + name = f"writer-{count % 85}" + if imports and count < 85: + comms.threads.import_thread( + ClaudeTranscript(root / "source.jsonl"), + name=name, + worktree=str(root), + tags=frozenset({"proof"}), + ) + else: + thread = ( + comms.registry.require(name) + if imports + else declaration(root, name, float(count % 85 + 1)) + ) + comms.registry.register(replace(thread, title=f"write {count}"), StoppedThreadStatus()) + count += 1 + result.put(count) + + +def main() -> None: + parser = argparse.ArgumentParser() + parser.add_argument("root", type=Path) + parser.add_argument("--seconds", type=float, default=120) + parser.add_argument("--imports", action="store_true") + parser.add_argument("--dump-after", type=float, default=0) + args = parser.parse_args() + if args.dump_after: + faulthandler.dump_traceback_later(args.dump_after) + root = args.root.resolve() + root.mkdir(mode=0o700, parents=True, exist_ok=True) + comms = Comms(root) + comms.messaging.initialize_private_initial_protocol() + event = dict( + timestamp="2026-10-19T00:00:00Z", + cwd=str(root), + sessionId="proof", + promptId="p-1", + origin=dict(kind="human"), + ) + with (root / "source.jsonl").open("w") as stream: + for kind, text in (("user", "Scratch input"), ("assistant", "Scratch reply")): + content = [dict(type="text", text=text)] + event.update(type=kind, uuid=kind, message=dict(role=kind, content=content)) + print(json.dumps(event), file=stream) + comms.registry.register(declaration(root, "sentinel"), StoppedThreadStatus()) + reading = ObserveSidebar(str(root), True, False).observation() + stop = mp.Event() + counts: Queue[int] = mp.Queue() + process = mp.Process(target=writer, args=(root, stop, counts, args.imports)) + process.start() + start = time.monotonic() + reads = empty = errors = 0 + try: + while time.monotonic() - start < args.seconds: + for result in reading.read(comms, comms.views.revision()): + if isinstance(result, SidebarObserved): + reads += 1 + if not result.derived[1]: + empty += 1 + print(f"EMPTY after {reads} reads", flush=True) + elif isinstance(result, SidebarUnreadable): + errors += 1 + print(f"UNREADABLE after {reads} reads: {result.error}", flush=True) + if not process.is_alive(): + raise RuntimeError(f"writer exited: {process.exitcode}") + finally: + stop.set() + process.join(15) + comms.views.transcript_reads.close() + print( + f"elapsed={time.monotonic()-start:.3f}s reads={reads} empty={empty} errors={errors} " + f"writes={counts.get(timeout=2)} writer_exit={process.exitcode} imports={args.imports}", + flush=True, + ) + if (reads > 0, empty, errors, process.exitcode) != (True, 0, 0, 0): + raise RuntimeError("Sidebar observation failed; see the original results above") + + +if __name__ == "__main__": + main()