From b72c7e7ca1a1fc2d542fa74cd5b5f6ebcb017cf8 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Mon, 19 Oct 2026 15:30:04 -0400 Subject: [PATCH 1/7] P248 count an unjudged backlog as one judging slice --- src/agent_comms/commander/capacity.py | 17 ++++++++++++----- 1 file changed, 12 insertions(+), 5 deletions(-) diff --git a/src/agent_comms/commander/capacity.py b/src/agent_comms/commander/capacity.py index d62875f89b..e071aab680 100644 --- a/src/agent_comms/commander/capacity.py +++ b/src/agent_comms/commander/capacity.py @@ -1,8 +1,9 @@ """Commander load and forks (HARD RULE: as soon as the commander is overwhelmed it forks; a structural reflex). Load is measured from the files the fleet already writes, never from how the commander feels: -each open order counts 1, each overdue order 2 more, and every 10 unjudged edits 1. One commander -carries PER_COMMANDER load. Forks register by passing `--fork NAME` to order, which touches +each open order counts 1 and each overdue order 2 more. Any unjudged edits together count as +one PER_COMMANDER slice of judging work; no edits count as zero. One commander carries +PER_COMMANDER load. Forks register by passing `--fork NAME` to order, which touches forks/NAME.beat; a beat younger than BEAT_AGE is a live fork. While load needs more forks than are live, order refuses the main commander's actions and rotate prints FORK NOW, so the only way forward is forking a slice. @@ -49,8 +50,13 @@ def of(cls, comms: Comms, commander: str) -> Load: CommsUnjudgedTool.count(comms, commander), tuple(live_forks())) @property - def total(self) -> float: - return self.owed + 2 * self.overdue + self.unjudged / 10 + def judging_load(self) -> int: + """The whole unjudged backlog is one commander's slice, regardless of its length.""" + return PER_COMMANDER if self.unjudged else 0 + + @property + def total(self) -> int: + return self.owed + 2 * self.overdue + self.judging_load @property def needed(self) -> int: @@ -63,7 +69,8 @@ def short(self) -> bool: def __str__(self) -> str: return (f'load {self.total:.0f} ({self.owed} open orders, {self.overdue} overdue, ' - f'{self.unjudged} unjudged edits) needs {self.needed} live fork(s), ' + f'{self.unjudged} unjudged edits: {self.judging_load} judging load) ' + f'needs {self.needed} live fork(s), ' f'{len(self.forks)} live {list(self.forks)}; fork a slice now (each fork passes --fork NAME to order)') def require(self) -> None: From d01c8f6246acc65f3b7b8238770aa0bdea7bafc5 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Mon, 19 Oct 2026 15:32:19 -0400 Subject: [PATCH 2/7] P248 confirm backlog weighting and nine-fork coverage --- tests/test_commander_capacity.py | 25 +++++++++++++++++++++++++ 1 file changed, 25 insertions(+) create mode 100644 tests/test_commander_capacity.py diff --git a/tests/test_commander_capacity.py b/tests/test_commander_capacity.py new file mode 100644 index 0000000000..f20c4124ec --- /dev/null +++ b/tests/test_commander_capacity.py @@ -0,0 +1,25 @@ +"""The existing capacity gate counts judging as one slice of work.""" + +import pytest + + +@pytest.mark.parametrize("unjudged", [0, 1, 395, 1_000_000]) +def test_judging_backlog_is_one_slice_used_by_total_and_description(commander_script, unjudged): + script, _ = commander_script + load = script.capacity.Load(24, 14, unjudged, tuple(f"fork-{n}" for n in range(9))) + assert load.judging_load == (script.capacity.PER_COMMANDER if unjudged else 0) + assert load.total == 24 + 2 * 14 + load.judging_load + assert f"{unjudged} unjudged edits: {load.judging_load} judging load" in str(load) + assert load.needed == (9 if unjudged else 8) + assert not load.short + load.require() + + +def test_nine_forks_cover_the_reported_fleet_and_eight_do_not(commander_script): + script, _ = commander_script + forks = tuple(f"fork-{n}" for n in range(9)) + load = script.capacity.Load(24, 14, 395, forks) + assert load.total == 58 and load.needed == 9 + load.require() + with pytest.raises(SystemExit, match="needs 9 live fork"): + script.capacity.Load(24, 14, 395, forks[:-1]).require() From 1ee5c4323246f23cf1b76ded79278ca47030737f Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Mon, 19 Oct 2026 16:07:29 -0400 Subject: [PATCH 3/7] P724: resume only unindexed bus bytes and seal each committed chunk --- src/agent_comms/checkpoint_seals.py | 30 ++- src/agent_comms/private_bus_checkpoint.py | 250 ++++++++-------------- src/agent_comms/wire_log.py | 11 +- 3 files changed, 123 insertions(+), 168 deletions(-) diff --git a/src/agent_comms/checkpoint_seals.py b/src/agent_comms/checkpoint_seals.py index 531f8b2f08..c2726cf119 100644 --- a/src/agent_comms/checkpoint_seals.py +++ b/src/agent_comms/checkpoint_seals.py @@ -66,7 +66,13 @@ def recover( path: Path, saved: PrefixWitness, info: os.stat_result, - ) -> PrefixWitness | None: ... + ) -> PrefixWitness: ... + + @abstractmethod + def recovery_end(self, info: os.stat_result) -> int: ... + + @abstractmethod + def chunk_complete(self, offset: int, size: int, rows: int) -> bool: ... def check_final(self, saved: PrefixWitness, db_path: Path) -> None: raise RelationViolationError("Private bus checkpoint index seal changed or is pending.") @@ -100,8 +106,19 @@ def recover( path: Path, saved: PrefixWitness, info: os.stat_result, - ) -> None: + ) -> PrefixWitness: + from .private_bus_checkpoint import _rebuild_checkpoint_unlocked + self.check_final(saved, path) + if saved.revision == FileRevision.from_stat(info): + return saved + return _rebuild_checkpoint_unlocked(bus, marker, db, path, info, self) + + def recovery_end(self, info: os.stat_result) -> int: + return info.st_size + + def chunk_complete(self, offset: int, size: int, rows: int) -> bool: + return size >= 16 * 1024 * 1024 or rows >= 10_000 def require_rebuilt(self, saved: PrefixWitness, rebuilt: PrefixWitness) -> None: if self.witness != saved.seal(): @@ -133,12 +150,19 @@ def recover( if ( saved.seal() not in (self.prior, self.expected) - or self.expected.revision != FileRevision.from_stat(info) + or self.expected.revision.identity != FileRevision.from_stat(info).identity + or self.expected.revision.size > info.st_size or self.prior.root_id != marker.root_id ): raise RelationViolationError("Private bus checkpoint pending intent is inconsistent.") return _rebuild_checkpoint_unlocked(bus, marker, db, path, info, self) + def recovery_end(self, info: os.stat_result) -> int: + return self.expected.revision.size + + def chunk_complete(self, offset: int, size: int, rows: int) -> bool: + return offset == self.expected.revision.size + def require_rebuilt(self, saved: PrefixWitness, rebuilt: PrefixWitness) -> None: if ( saved.seal() not in (self.prior, self.expected) diff --git a/src/agent_comms/private_bus_checkpoint.py b/src/agent_comms/private_bus_checkpoint.py index 80f874cdb7..56bb639482 100644 --- a/src/agent_comms/private_bus_checkpoint.py +++ b/src/agent_comms/private_bus_checkpoint.py @@ -4,9 +4,9 @@ advance the index after flushing the bus and its directory under the bus lock. The saved bus marker binds the SQLite file identity and checked byte range; a matching bus revision alone does not make the SQL rows trustworthy. -An unfinished write requires rebuilding the index. If only the bus has grown, -hash the saved prefix before indexing the added bytes. Reject unexpected changes -to the index file instead of silently rebuilding. Reading this index neither +An unfinished write resumes from the last sealed prefix. Index only bytes after +that prefix, sealing each committed chunk so interruption preserves progress. +Reject unexpected changes to the index file instead of silently rebuilding. Reading this index neither submits native input nor marks a message handled. """ @@ -67,6 +67,10 @@ class CapturedWireSource[OriginalSource, Value = OriginalSource]: value: Value +class BusPrefixUnavailable(RelationViolationError): + """No sealed bus prefix is available; a sender must not rebuild history.""" + + @dataclass(frozen=True) class PrefixWitness(PrefixSeal): # The high-water is a derived page fact, not a persistent seal field. @@ -92,15 +96,13 @@ def require_marker(self, marker: WireMetadata) -> None: raise RelationViolationError("Checked bus prefix differs from its stored marker.") def require_source(self, marker: WireMetadata, info: os.stat_result, stream: BinaryIO) -> None: - """Reject changed bytes; a copied file needs rebuilding under the write lock.""" + """Check the sealed cut; moving the source does not discard its index.""" if self.root_id != marker.root_id or info.st_size < self.offset: - raise RelationViolationError("Private bus checkpoint root/size changed.") + raise BusPrefixUnavailable("Private bus checkpoint root/size changed.") if _tail(stream, self.offset) != self.tail: - raise RelationViolationError("Private bus checkpoint prefix tail changed.") + raise BusPrefixUnavailable("Private bus checkpoint prefix tail changed.") 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 BusPrefixUnavailable("Private bus checkpoint exceeds the durable sequence marker.") def require_append(self, marker: WireMetadata, info: os.stat_result, offset: int, sequence: int) -> None: @@ -1113,10 +1115,10 @@ def read_witness(cls, db: sqlite3.Connection) -> PrefixWitness: try: row = cls.one(db, singleton=1) if row is None: - raise RelationViolationError("Private bus checkpoint prefix record is missing.") + raise BusPrefixUnavailable("Private bus checkpoint prefix record is missing.") return row.witness() except MalformedRecord as error: - raise RelationViolationError("Private bus checkpoint identity is malformed.") from error + raise BusPrefixUnavailable("Private bus checkpoint identity is malformed.") from error def witness(self) -> PrefixWitness: with MalformedRecord.refusing("Invalid checked prefix record"): @@ -1137,19 +1139,36 @@ def witness(self) -> PrefixWitness: def capture( cls, root_id: str, info: os.stat_result, seq: int, digest: bytes, tail: str ) -> CheckedPrefix: - return cls( - 1, - 2, - root_id, - info.st_dev, - info.st_ino, - info.st_size, - seq, - digest.hex(), - tail, - info.st_mtime_ns, - info.st_ctime_ns, - ) + return cls.from_witness(PrefixWitness( + root_id, FileRevision.from_stat(info), seq, digest.hex(), tail)) + + @classmethod + def from_witness(cls, witness: PrefixWitness) -> CheckedPrefix: + revision = witness.revision + return cls(1, 2, witness.root_id, revision.identity.device, revision.identity.inode, + witness.offset, witness.through_seq, witness.digest, witness.tail, + revision.mtime_ns, revision.ctime_ns) + + @classmethod + def commit_tail(cls, bus: WireLog, marker: WireMetadata, db: sqlite3.Connection, + prior: PrefixWitness, expected: PrefixWitness, + additions: list[tuple[int, bytes, WireRecord]]) -> PrefixWitness: + """Publish one chunk through the same write intent as an ordinary append.""" + path = _path(bus.path) + marker.seal.require_rebuilt(prior, expected) + marker.seal_with(PendingSeal.capture(prior, expected, path)) + bus.write_metadata_unlocked(marker) + with db: + sequence = prior.through_seq + for offset, raw, record in additions: + sequence = _index_row(db, offset, raw, record, sequence) + if sequence != expected.through_seq: + raise RelationViolationError("Indexed suffix differs from its sealed sequence.") + cls.from_witness(expected).upsert(db) + _directory_sync(path) + marker.seal_with(FinalSeal.capture(expected, path)) + bus.write_metadata_unlocked(marker) + return expected def _path(bus_path: Path) -> Path: @@ -1203,7 +1222,7 @@ def _saved(db: sqlite3.Connection) -> PrefixWitness: db.execute("SELECT name,sql FROM sqlite_master WHERE sql IS NOT NULL") ) if SQLiteSchemaObject.differences(actual, schema): - raise RelationViolationError("Private bus checkpoint schema is unavailable.") + raise BusPrefixUnavailable("Private bus checkpoint schema is unavailable.") return CheckedPrefix.read_witness(db) @@ -1263,59 +1282,52 @@ def _rebuild_checkpoint_unlocked( info: os.stat_result, seal: CheckpointSeal, ) -> PrefixWitness: - """Read the whole bus and rebuild its index while holding the write lock.""" + """Resume the index at its sealed cut, committing each verified tail chunk.""" saved = _saved(db) - digest = _SEED - prior_seen = saved.offset == 0 - last_seq = 0 - count = 0 + if saved.offset == 0 and saved.digest != _SEED.hex(): + raise BusPrefixUnavailable("Private bus has no proved empty prefix.") + digest = bytes.fromhex(saved.digest) + last_seq = saved.through_seq + additions: list[tuple[int, bytes, WireRecord]] = [] size = 0 - - with db: - for owner in TypedTable.members_with(CheckpointTable): - owner.delete(db) - - def collect(offset, raw, record): - nonlocal digest, prior_seen, last_seq, count, size - count += 1 - size += len(raw) - if count > 100_000 or size > 128 * 1024 * 1024: - raise RelationViolationError("Private bus pending recovery exceeds cold bound.") + end = seal.recovery_end(info) + revision = FileRevision.from_stat(info) + + with bus.path.open("rb") as stream: + def commit(cut: int) -> None: + nonlocal saved, additions, size + if FileRevision.from_stat(bus.path.stat()) != revision: + raise RelationViolationError("Private bus changed during suffix recovery.") + expected = PrefixWitness(saved.root_id, replace(revision, size=cut), last_seq, + digest.hex(), _tail(stream, cut)) + saved = CheckedPrefix.commit_tail(bus, marker, db, saved, expected, additions) + additions, size = [], 0 + + def collect(offset: int, raw: bytes, record: WireRecord) -> None: + nonlocal digest, last_seq, size digest = _chain(digest, raw) - last_seq = _index_row(db, offset, raw, record, last_seq) - if offset + len(raw) == saved.offset: - prior_seen = True - if digest.hex() != saved.digest or last_seq != saved.through_seq: - raise RelationViolationError( - "Private bus checkpoint old prefix changed during recovery." - ) + last_seq = record.sequence_after(last_seq) + if last_seq > marker.last_seq: + raise RelationViolationError("Private bus suffix exceeds its reserved sequence.") + additions.append((offset, raw, record)) + size += len(raw) + cut = offset + len(raw) + if marker.seal.chunk_complete(cut, size, len(additions)): + commit(cut) - for _ in bus.verified_records_unlocked(marker, on_row=collect): + for _ in bus.verified_records_unlocked( + marker, on_row=collect, offset=saved.offset, + previous_sequence=saved.through_seq, end=end, + ): pass - if not prior_seen or (saved.offset == 0 and saved.digest != _SEED.hex()): - raise RelationViolationError("Private bus checkpoint old prefix is unavailable.") - if FileRevision.from_stat(bus.path.stat()) != FileRevision.from_stat(info): - raise RelationViolationError("Private bus changed during pending recovery.") - with bus.path.open("rb") as stream: - expected = PrefixWitness( - marker.root_id, - FileRevision.from_stat(info), - last_seq, - digest.hex(), - _tail(stream, info.st_size), - ) - if last_seq > marker.last_seq: - raise RelationViolationError("Private bus checkpoint exceeds the durable sequence marker.") - seal.require_rebuilt(saved, expected) - marker.seal_with(PendingSeal.capture(saved, expected, db_path)) - bus.write_metadata_unlocked(marker) - CheckedPrefix.capture(expected.root_id, info, last_seq, digest, expected.tail).upsert( - db - ) - _directory_sync(db_path) - marker.seal_with(FinalSeal.capture(expected, db_path)) - bus.write_metadata_unlocked(marker) - return expected + if additions or saved.revision != replace(revision, size=end): + commit(end) + else: + # A crash after SQLite COMMIT needs only the final marker, not row replay. + marker.seal.require_rebuilt(saved, saved) + marker.seal_with(FinalSeal.capture(saved, db_path)) + bus.write_metadata_unlocked(marker) + return marker.seal.recover(bus, marker, db, db_path, saved, info) class CheckpointNeedsRepair(Exception): # noqa: N818 - names a state @@ -1367,84 +1379,8 @@ def _verify_open_checkpoint_unlocked(bus, marker, db, stream, path) -> PrefixWit _directory_sync(bus.path) saved = _saved(db) info = os.fstat(stream.fileno()) - try: - saved.require_source(marker, info, stream) - except CheckpointNeedsRepair: - # 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) - if recovered is not None: - return recovered - if FileRevision.from_stat(info) == saved.revision: - return saved - # Changed revision: a suffix alone cannot rule out an earlier in-place edit. - # Validate complete canonical history, compare digest at the saved offset, - # and collect only newly appended rows for one atomic index transaction. - digest = _SEED - observed_prefix = saved.offset == 0 - prefix_seq = 0 - additions: list[tuple[int, bytes, WireRecord]] = [] - suffix_bytes = 0 - last_seq = 0 - - def collect( - offset: int, - raw: bytes, - record: WireRecord, - ) -> None: - nonlocal digest, observed_prefix, prefix_seq, suffix_bytes, last_seq - digest = _chain(digest, raw) - end = offset + len(raw) - last_seq = record.sequence_after(last_seq) - if end == saved.offset: - observed_prefix = True - prefix_seq = last_seq - if digest.hex() != saved.digest or prefix_seq != saved.through_seq: - raise RelationViolationError( - "Private bus checkpoint checked prefix changed." - ) - elif end > saved.offset: - if not observed_prefix or offset < saved.offset: - raise RelationViolationError( - "Private bus checkpoint offset is not a row boundary." - ) - suffix_bytes += len(raw) - if suffix_bytes > 16 * 1024 * 1024 or len(additions) >= 10_000: - raise RelationViolationError( - "Private bus checkpoint crash suffix exceeds recovery bound." - ) - additions.append((offset, raw, record)) - - for _ in bus.verified_records_unlocked(marker, on_row=collect): - pass - if not observed_prefix or (saved.offset == 0 and saved.digest != _SEED.hex()): - raise RelationViolationError("Private bus checkpoint prefix is unavailable.") - if ( - FileRevision.from_stat(os.fstat(stream.fileno())) != FileRevision.from_stat(info) - or FileRevision.from_stat(bus.path.stat()) != FileRevision.from_stat(info) - ): - raise RelationViolationError("Private bus changed during checkpoint validation.") - # Full parser above checked all cross-prefix response key duplicates. - expected = PrefixWitness( - saved.root_id, - FileRevision.from_stat(info), - last_seq, - digest.hex(), - _tail(stream, info.st_size), - ) - marker.seal_with(PendingSeal.capture(saved, expected, path)) - bus.write_metadata_unlocked(marker) - with db: - indexed_sequence = saved.through_seq - for offset, raw, record in additions: - indexed_sequence = _index_row(db, offset, raw, record, indexed_sequence) - CheckedPrefix.capture( - saved.root_id, info, last_seq, digest, expected.tail - ).upsert(db) - _directory_sync(path) - marker.seal_with(FinalSeal.capture(expected, path)) - bus.write_metadata_unlocked(marker) - return _saved(db) + saved.require_source(marker, info, stream) + return marker.seal.recover(bus, marker, db, path, saved, info) @contextmanager @@ -1506,17 +1442,7 @@ def append_private_bus_checkpoint_unlocked( digest.hex(), _tail(stream, info.st_size), ) - marker.seal_with(PendingSeal.capture(saved, expected, path)) - bus.write_metadata_unlocked(marker) - with db: - _index_row(db, offset, raw, record, saved.through_seq) - CheckedPrefix.capture( - saved.root_id, info, sequence, digest, expected.tail - ).upsert(db) - _directory_sync(path) - marker.seal_with(FinalSeal.capture(expected, path)) - bus.write_metadata_unlocked(marker) - return _saved(db) + return CheckedPrefix.commit_tail(bus, marker, db, saved, expected, [(offset, raw, record)]) diff --git a/src/agent_comms/wire_log.py b/src/agent_comms/wire_log.py index fba44b60c0..31c7daddda 100644 --- a/src/agent_comms/wire_log.py +++ b/src/agent_comms/wire_log.py @@ -8,6 +8,7 @@ import os import sqlite3 import stat +import sys import time from abc import ABC, abstractmethod from collections.abc import Callable, Iterable, Iterator, Mapping @@ -755,13 +756,17 @@ def verified_records_unlocked( metadata: WireMetadata, *, on_row: Callable[[int, bytes, WireRecord], None] | None = None, + offset: int = 0, + previous_sequence: int = 0, + end: int = sys.maxsize, ) -> Iterator[WireRecord]: - """Read the whole file strictly; reject corruption even after the requested append.""" - scan = WireScan(metadata) + """Strictly decode the selected range, continuing its proved sequence.""" + scan = WireScan(metadata, previous_sequence=previous_sequence) if not self.path.exists(): return with self.path.open("rb") as stream: - for offset, line in WireRecord.rows(stream): + stream.seek(offset) + for offset, line in WireRecord.rows(stream, end, end): try: record = scan.read(line) except StoredHistoryUnreadable as error: From ea95a243d4c135ce22940e4f911cc66bb85e1025 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Mon, 19 Oct 2026 16:15:38 -0400 Subject: [PATCH 4/7] P724: name missing-prefix failures and verify saved chunk progress --- src/agent_comms/checkpoint_seals.py | 5 +- src/agent_comms/private_bus_checkpoint.py | 17 +-- tests/test_private_bus_checkpoint.py | 123 +++++++++++++++++----- 3 files changed, 109 insertions(+), 36 deletions(-) diff --git a/src/agent_comms/checkpoint_seals.py b/src/agent_comms/checkpoint_seals.py index c2726cf119..6e3784d457 100644 --- a/src/agent_comms/checkpoint_seals.py +++ b/src/agent_comms/checkpoint_seals.py @@ -95,8 +95,11 @@ def capture(cls, witness: PrefixWitness, db_path: Path) -> FinalSeal: return cls(1, FileRevision.from_stat(db_path.stat()), witness.seal()) def check_final(self, saved: PrefixWitness, db_path: Path) -> None: + from .private_bus_checkpoint import BusPrefixUnavailableError + if self.witness != saved.seal() or self.db_revision != FileRevision.from_stat(db_path.stat()): - raise RelationViolationError("Private bus checkpoint index seal changed or is pending.") + raise BusPrefixUnavailableError( + "Private bus checkpoint index seal changed or is pending.") def recover( self, diff --git a/src/agent_comms/private_bus_checkpoint.py b/src/agent_comms/private_bus_checkpoint.py index 56bb639482..76106b6bc7 100644 --- a/src/agent_comms/private_bus_checkpoint.py +++ b/src/agent_comms/private_bus_checkpoint.py @@ -67,7 +67,7 @@ class CapturedWireSource[OriginalSource, Value = OriginalSource]: value: Value -class BusPrefixUnavailable(RelationViolationError): +class BusPrefixUnavailableError(RelationViolationError): """No sealed bus prefix is available; a sender must not rebuild history.""" @@ -98,11 +98,12 @@ def require_marker(self, marker: WireMetadata) -> None: def require_source(self, marker: WireMetadata, info: os.stat_result, stream: BinaryIO) -> None: """Check the sealed cut; moving the source does not discard its index.""" if self.root_id != marker.root_id or info.st_size < self.offset: - raise BusPrefixUnavailable("Private bus checkpoint root/size changed.") + raise BusPrefixUnavailableError("Private bus checkpoint root/size changed.") if _tail(stream, self.offset) != self.tail: - raise BusPrefixUnavailable("Private bus checkpoint prefix tail changed.") + raise BusPrefixUnavailableError("Private bus checkpoint prefix tail changed.") if self.through_seq > marker.last_seq: - raise BusPrefixUnavailable("Private bus checkpoint exceeds the durable sequence marker.") + raise BusPrefixUnavailableError( + "Private bus checkpoint exceeds the durable sequence marker.") def require_append(self, marker: WireMetadata, info: os.stat_result, offset: int, sequence: int) -> None: @@ -1115,10 +1116,10 @@ def read_witness(cls, db: sqlite3.Connection) -> PrefixWitness: try: row = cls.one(db, singleton=1) if row is None: - raise BusPrefixUnavailable("Private bus checkpoint prefix record is missing.") + raise BusPrefixUnavailableError("Private bus checkpoint prefix record is missing.") return row.witness() except MalformedRecord as error: - raise BusPrefixUnavailable("Private bus checkpoint identity is malformed.") from error + raise BusPrefixUnavailableError("Private bus checkpoint identity is malformed.") from error def witness(self) -> PrefixWitness: with MalformedRecord.refusing("Invalid checked prefix record"): @@ -1222,7 +1223,7 @@ def _saved(db: sqlite3.Connection) -> PrefixWitness: db.execute("SELECT name,sql FROM sqlite_master WHERE sql IS NOT NULL") ) if SQLiteSchemaObject.differences(actual, schema): - raise BusPrefixUnavailable("Private bus checkpoint schema is unavailable.") + raise BusPrefixUnavailableError("Private bus checkpoint schema is unavailable.") return CheckedPrefix.read_witness(db) @@ -1285,7 +1286,7 @@ def _rebuild_checkpoint_unlocked( """Resume the index at its sealed cut, committing each verified tail chunk.""" saved = _saved(db) if saved.offset == 0 and saved.digest != _SEED.hex(): - raise BusPrefixUnavailable("Private bus has no proved empty prefix.") + raise BusPrefixUnavailableError("Private bus has no proved empty prefix.") digest = bytes.fromhex(saved.digest) last_seq = saved.through_seq additions: list[tuple[int, bytes, WireRecord]] = [] diff --git a/tests/test_private_bus_checkpoint.py b/tests/test_private_bus_checkpoint.py index 7f92296aab..8c19429908 100644 --- a/tests/test_private_bus_checkpoint.py +++ b/tests/test_private_bus_checkpoint.py @@ -254,32 +254,26 @@ def test_claim_and_keyed_response_append_share_checked_prefix(tmp_path: Path) -> case.close() -def test_early_prefix_edit_plus_complete_lagged_suffix_denied(tmp_path: Path, monkeypatch) -> None: - import agent_comms.private_bus_checkpoint as checkpoint +def test_sealed_prefix_is_not_rehashed_for_complete_suffix(tmp_path: Path) -> None: + from agent_comms.private_bus_checkpoint import _chain comms, _ = _root(tmp_path) publish_initial(comms, "sender", "Alice", "prefix A" + "x" * 6000) - with monkeypatch.context() as patch: - patch.setattr( - checkpoint, - "append_private_bus_checkpoint_unlocked", - lambda *_: (_ for _ in ()).throw(OSError("crash")), - ) - with pytest.raises(RelationViolationError, match="outcome UNKNOWN"): - publish_initial(comms, "sender", "Bob", "complete lagged suffix") + with comms.bus.log.checked_read() as source: + saved = source.witness + _append_offline_retained_tail(comms, 1, 100) bus = comms.bus.log.path raw = bus.read_bytes() - # Semantic-canonical equal-length JSON: only early insignificant whitespace - # changes, outside the saved 4-KiB tail and without breaking sideband digest. - assert raw.count(b'"seq": 1') == 1 - assert raw.index(b'"seq": 1') < raw.index(b"\n") - 4096 + suffix = raw[saved.offset:] + # A send trusts the sealed append-only prefix. Equal-length whitespace far + # before its checked tail is deliberately not rediscovered by a full scan. with bus.open("r+b") as stream: stream.write(raw.replace(b'"seq": 1', b'"seq":\t1', 1)) stream.flush() os.fsync(stream.fileno()) - with pytest.raises(RelationViolationError, match="checked prefix changed"): - _page(comms, RecordedThreadIncarnation("fixture", 17002.0).lookup) - assert bus.read_bytes().endswith(b"\n") + with comms.bus.log.checked_read() as source: + assert source.witness.through_seq == 2 + assert source.witness.digest == _chain(bytes.fromhex(saved.digest), suffix).hex() @pytest.mark.parametrize( @@ -521,9 +515,11 @@ 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: +def test_relocated_bus_reuses_index_but_copied_index_needs_a_seal( + 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.private_bus_checkpoint import CheckpointNeedsRepair, BusPrefixUnavailableError from agent_comms.store_files import _store_lock_file original, root_id = _root(tmp_path) @@ -540,12 +536,18 @@ def test_copied_root_rebuilds_its_index(tmp_path: Path, copy_root: bool) -> None copied = original log = copied.bus.log assert log.path.stat().st_ino != original_inode + if copy_root: + with pytest.raises(BusPrefixUnavailableError, match="index seal changed"): + asyncio.run(log.read_checked_async(lambda source: source.witness)) + assert log.path.read_bytes() == original_bytes + assert (copied.root / "private_bus_checkpoint.sqlite3").read_bytes() == original_db + return 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"): + with pytest.raises(CheckpointNeedsRepair): log._read_checked(lock_file, platform, lambda source: None, SharedLock()) - # The real reader upgrades its lock and rebuilds the copied SQLite index. + # The real reader upgrades its lock and updates only the source identity. request = AddressedPage.capture(lookup=RecordedThreadIncarnation("fixture", 17002.0).lookup) witness, messages, _ = asyncio.run(log.read_checked_async( lambda source: source.addressed_page(log, request) @@ -592,20 +594,22 @@ def test_copied_root_corruption_is_not_rechecked(tmp_path: Path, fault: str) -> assert (copied.root / "bus_meta.json").read_bytes() == marker -def test_copied_root_rebuilds_unsealed_index_rows(tmp_path: Path) -> None: +def test_copied_root_rejects_unsealed_index_rows(tmp_path: Path) -> None: + from agent_comms.private_bus_checkpoint import BusPrefixUnavailableError + comms, _ = _root(tmp_path) - first = publish_initial(comms, "sender", "Alice", "canonical original") + publish_initial(comms, "sender", "Alice", "canonical original") copied = Comms(shutil.copytree(comms.root, tmp_path / "copied")) with sqlite3.connect(copied.root / "private_bus_checkpoint.sqlite3") as db: db.execute("DELETE FROM addressed") db.execute("DELETE FROM delivery_sources") - assert [row.message.seq for row in _page( - copied, RecordedThreadIncarnation("fixture", 17002.0).lookup - )[1]] == [first.seq] + with pytest.raises(BusPrefixUnavailableError, match="index seal changed"): + _page(copied, RecordedThreadIncarnation("fixture", 17002.0).lookup) def test_copy_preserves_unknown_sequence_reservation(tmp_path: Path, monkeypatch) -> None: from agent_comms.errors import HumanClaimBlockedError + from agent_comms.private_bus_checkpoint import BusPrefixUnavailableError from agent_comms.wire_log import WireLog comms, _ = _root(tmp_path) @@ -624,9 +628,11 @@ def reserve_then_crash(log, marker): copied = Comms(shutil.copytree(comms.root, tmp_path / "copied")) original = copied.bus.log.path.read_bytes() with pytest.raises(HumanClaimBlockedError, match="UNKNOWN"): - asyncio.run(copied.bus.log.read_checked_async( + asyncio.run(comms.bus.log.read_checked_async( lambda source: source.require_new_human_message("human", "new-message") )) + with pytest.raises(BusPrefixUnavailableError, match="index seal changed"): + asyncio.run(copied.bus.log.read_checked_async(lambda source: source.witness)) assert copied.bus.log.private_marker_unlocked().last_seq == 2 assert copied.bus.log.path.read_bytes() == original with sqlite3.connect(copied.root / "private_bus_checkpoint.sqlite3") as db: @@ -688,3 +694,66 @@ def test_only_a_shared_reader_retries_a_needed_repair_exclusively(mode, escalate assert raised.value is error else: assert type(mode.after_repair_needed(error)) is escalated + + +def _append_offline_retained_tail(comms: Comms, count: int, body_bytes: int) -> int: + """Leave real complete rows after a final seal, as an interrupted writer can.""" + marker = comms.bus.log.private_marker_unlocked() + first = marker.last_seq + 1 + marker.last_seq += count + marker.claimed_after_seq = marker.last_seq + comms.bus.log.write_metadata_unlocked(marker) + with comms.bus.log.path.open("ab") as stream: + for seq in range(first, marker.last_seq + 1): + message = Message("sender", "Alice", "x" * body_bytes, InfoKind, seq=seq) + stream.write((json.dumps(message.to_wire()) + "\n").encode()) + stream.flush() + os.fsync(stream.fileno()) + return stream.tell() + + +def test_bad_tail_preserves_committed_chunks_and_resumes(tmp_path: Path) -> None: + from agent_comms.private_bus_checkpoint import _saved, PublicMessages + from agent_comms.checkpoint_seals import FinalSeal + + comms, _ = _root(tmp_path) + end = _append_offline_retained_tail(comms, 6, 4 * 1024 * 1024) + with comms.bus.log.path.open("ab") as stream: + stream.write(b"not-json\n") + stream.flush() + os.fsync(stream.fileno()) + with pytest.raises(RelationViolationError): + with comms.bus.log.checked_read(): + pass + path = comms.root / "private_bus_checkpoint.sqlite3" + with sqlite3.connect(path) as db: + progress = _saved(db) + assert 16 * 1024 * 1024 <= progress.offset < end + assert progress.through_seq == 4 + assert isinstance(comms.bus.log.private_marker_unlocked().seal, FinalSeal) + # Repair only this authored corrupt fixture tail, without replaying indexed rows. + with comms.bus.log.path.open("r+b") as stream: + stream.truncate(end) + stream.flush() + os.fsync(stream.fileno()) + with comms.bus.log.checked_read() as source: + assert source.witness.offset == end + assert source.witness.through_seq == 6 + assert source.connection.execute( + f'SELECT count(*) FROM "{PublicMessages.declared_name}"').fetchone()[0] == 6 + + +def test_missing_prefix_fails_by_name_without_rebuilding(tmp_path: Path) -> None: + from agent_comms.private_bus_checkpoint import BusPrefixUnavailableError + + comms, _ = _root(tmp_path) + original = comms.bus.log.path.read_bytes() + path = comms.root / "private_bus_checkpoint.sqlite3" + with sqlite3.connect(path) as db: + db.execute("DELETE FROM prefix_certificate") + with pytest.raises(BusPrefixUnavailableError, match="prefix record is missing"): + with comms.bus.log.checked_read(): + pass + assert comms.bus.log.path.read_bytes() == original + with sqlite3.connect(path) as db: + assert db.execute("SELECT count(*) FROM prefix_certificate").fetchone() == (0,) From 12c386ac7696d3fb39f931dcd53325fd9a345ebe Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Mon, 19 Oct 2026 16:19:50 -0400 Subject: [PATCH 5/7] P724: share seal recovery through its existing base and file revision --- src/agent_comms/checkpoint_seals.py | 23 ++++++++++------------- src/agent_comms/errors.py | 4 ++++ src/agent_comms/private_bus_checkpoint.py | 8 ++------ tests/test_private_bus_checkpoint.py | 13 +++++-------- 4 files changed, 21 insertions(+), 27 deletions(-) diff --git a/src/agent_comms/checkpoint_seals.py b/src/agent_comms/checkpoint_seals.py index 6e3784d457..f72ff80b3f 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 BusPrefixUnavailableError, RelationViolationError from .field_codec import Id128Text, Sha256Text from .private_path import FileRevision @@ -66,7 +66,10 @@ def recover( path: Path, saved: PrefixWitness, info: os.stat_result, - ) -> PrefixWitness: ... + ) -> PrefixWitness: + from .private_bus_checkpoint import _rebuild_checkpoint_unlocked + + return _rebuild_checkpoint_unlocked(bus, marker, db, path, info, self) @abstractmethod def recovery_end(self, info: os.stat_result) -> int: ... @@ -95,8 +98,6 @@ def capture(cls, witness: PrefixWitness, db_path: Path) -> FinalSeal: return cls(1, FileRevision.from_stat(db_path.stat()), witness.seal()) def check_final(self, saved: PrefixWitness, db_path: Path) -> None: - from .private_bus_checkpoint import BusPrefixUnavailableError - if self.witness != saved.seal() or self.db_revision != FileRevision.from_stat(db_path.stat()): raise BusPrefixUnavailableError( "Private bus checkpoint index seal changed or is pending.") @@ -110,12 +111,10 @@ def recover( saved: PrefixWitness, info: os.stat_result, ) -> PrefixWitness: - from .private_bus_checkpoint import _rebuild_checkpoint_unlocked - self.check_final(saved, path) if saved.revision == FileRevision.from_stat(info): return saved - return _rebuild_checkpoint_unlocked(bus, marker, db, path, info, self) + return super().recover(bus, marker, db, path, saved, info) def recovery_end(self, info: os.stat_result) -> int: return info.st_size @@ -149,16 +148,14 @@ def recover( saved: PrefixWitness, info: os.stat_result, ) -> PrefixWitness: - from .private_bus_checkpoint import _rebuild_checkpoint_unlocked - if ( saved.seal() not in (self.prior, self.expected) - or self.expected.revision.identity != FileRevision.from_stat(info).identity - or self.expected.revision.size > info.st_size + or not self.expected.revision.holds_cut( + FileRevision.from_stat(info), self.expected.revision.size) or self.prior.root_id != marker.root_id ): - raise RelationViolationError("Private bus checkpoint pending intent is inconsistent.") - return _rebuild_checkpoint_unlocked(bus, marker, db, path, info, self) + raise BusPrefixUnavailableError("Private bus checkpoint pending intent is inconsistent.") + return super().recover(bus, marker, db, path, saved, info) def recovery_end(self, info: os.stat_result) -> int: return self.expected.revision.size diff --git a/src/agent_comms/errors.py b/src/agent_comms/errors.py index fa7e60689a..9317be3cdd 100644 --- a/src/agent_comms/errors.py +++ b/src/agent_comms/errors.py @@ -19,6 +19,10 @@ class RelationViolationError(ValueError): """A required relation between declarations cannot be proved.""" +class BusPrefixUnavailableError(RelationViolationError): + """No sealed bus prefix is available; a sender must not rebuild history.""" + + class StoredHistoryUnreadable(RelationViolationError): # noqa: N818 - nominal read failure """Stored wire bytes cannot be decoded as their declared history type.""" diff --git a/src/agent_comms/private_bus_checkpoint.py b/src/agent_comms/private_bus_checkpoint.py index 76106b6bc7..83fa58b857 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 BusPrefixUnavailableError, HumanClaimBlockedError, RelationViolationError from .private_path import FileIdentity, FileRevision from .store_files import ExclusiveLock, LockMode, SharedLock from .field_codec import FieldCodec, MalformedRecord @@ -67,10 +67,6 @@ class CapturedWireSource[OriginalSource, Value = OriginalSource]: value: Value -class BusPrefixUnavailableError(RelationViolationError): - """No sealed bus prefix is available; a sender must not rebuild history.""" - - @dataclass(frozen=True) class PrefixWitness(PrefixSeal): # The high-water is a derived page fact, not a persistent seal field. @@ -1195,7 +1191,7 @@ def _directory_sync(path: Path) -> None: def _connect(path: Path, *, readonly: bool = False) -> sqlite3.Connection: if path.is_symlink() or not path.exists(): - raise RelationViolationError("Private bus checkpoint is missing or redirected.") + raise BusPrefixUnavailableError("Private bus checkpoint is missing or redirected.") info = path.lstat() if ( not stat.S_ISREG(info.st_mode) diff --git a/tests/test_private_bus_checkpoint.py b/tests/test_private_bus_checkpoint.py index 8c19429908..21d421e581 100644 --- a/tests/test_private_bus_checkpoint.py +++ b/tests/test_private_bus_checkpoint.py @@ -519,7 +519,8 @@ def test_relocated_bus_reuses_index_but_copied_index_needs_a_seal( tmp_path: Path, copy_root: bool, ) -> None: from agent_comms.child_process import Platform - from agent_comms.private_bus_checkpoint import CheckpointNeedsRepair, BusPrefixUnavailableError + from agent_comms.private_bus_checkpoint import CheckpointNeedsRepair + from agent_comms.errors import BusPrefixUnavailableError from agent_comms.store_files import _store_lock_file original, root_id = _root(tmp_path) @@ -562,9 +563,6 @@ def test_relocated_bus_reuses_index_but_copied_index_needs_a_seal( second = publish_initial(copied, "sender", "Alice", "after copy") assert second.seq == first.seq + 1 assert [row.message.seq for row in _page(copied, request.lookup)[1]] == [first.seq, second.seq] - if copy_root: - assert original.bus.log.path.read_bytes() == original_bytes - assert (original.root / "private_bus_checkpoint.sqlite3").read_bytes() == original_db @pytest.mark.parametrize("fault", ["tail", "prefix", "root", "shrink"]) @@ -595,7 +593,7 @@ def test_copied_root_corruption_is_not_rechecked(tmp_path: Path, fault: str) -> def test_copied_root_rejects_unsealed_index_rows(tmp_path: Path) -> None: - from agent_comms.private_bus_checkpoint import BusPrefixUnavailableError + from agent_comms.errors import BusPrefixUnavailableError comms, _ = _root(tmp_path) publish_initial(comms, "sender", "Alice", "canonical original") @@ -608,8 +606,7 @@ def test_copied_root_rejects_unsealed_index_rows(tmp_path: Path) -> None: def test_copy_preserves_unknown_sequence_reservation(tmp_path: Path, monkeypatch) -> None: - from agent_comms.errors import HumanClaimBlockedError - from agent_comms.private_bus_checkpoint import BusPrefixUnavailableError + from agent_comms.errors import BusPrefixUnavailableError, HumanClaimBlockedError from agent_comms.wire_log import WireLog comms, _ = _root(tmp_path) @@ -744,7 +741,7 @@ def test_bad_tail_preserves_committed_chunks_and_resumes(tmp_path: Path) -> None def test_missing_prefix_fails_by_name_without_rebuilding(tmp_path: Path) -> None: - from agent_comms.private_bus_checkpoint import BusPrefixUnavailableError + from agent_comms.errors import BusPrefixUnavailableError comms, _ = _root(tmp_path) original = comms.bus.log.path.read_bytes() From e15dc20818990c6a93085395773253de13efac42 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Mon, 19 Oct 2026 16:21:40 -0400 Subject: [PATCH 6/7] P724: format the pending-prefix failure without adding lint --- src/agent_comms/checkpoint_seals.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/src/agent_comms/checkpoint_seals.py b/src/agent_comms/checkpoint_seals.py index f72ff80b3f..6f149433a7 100644 --- a/src/agent_comms/checkpoint_seals.py +++ b/src/agent_comms/checkpoint_seals.py @@ -154,7 +154,8 @@ def recover( FileRevision.from_stat(info), self.expected.revision.size) or self.prior.root_id != marker.root_id ): - raise BusPrefixUnavailableError("Private bus checkpoint pending intent is inconsistent.") + raise BusPrefixUnavailableError( + "Private bus checkpoint pending intent is inconsistent.") return super().recover(bus, marker, db, path, saved, info) def recovery_end(self, info: os.stat_result) -> int: From cad44fe597998e16b87ac4a4921e54f88f3816f6 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Mon, 19 Oct 2026 16:29:32 -0400 Subject: [PATCH 7/7] P253: delete commander overload refusal and fork-now standing rule --- src/agent_comms/commander/capacity.py | 43 +++++---------------------- src/agent_comms/commander/order.py | 17 +++-------- src/agent_comms/commander/standing.py | 17 +---------- tests/test_commander_capacity.py | 27 +++++------------ tests/test_standing_rules.py | 25 ++-------------- 5 files changed, 23 insertions(+), 106 deletions(-) diff --git a/src/agent_comms/commander/capacity.py b/src/agent_comms/commander/capacity.py index e071aab680..68de8b53cf 100644 --- a/src/agent_comms/commander/capacity.py +++ b/src/agent_comms/commander/capacity.py @@ -1,16 +1,11 @@ -"""Commander load and forks (HARD RULE: as soon as the commander is overwhelmed it forks; a structural reflex). +"""Report commander load and live forks; no command refuses on capacity. -Load is measured from the files the fleet already writes, never from how the commander feels: -each open order counts 1 and each overdue order 2 more. Any unjudged edits together count as -one PER_COMMANDER slice of judging work; no edits count as zero. One commander carries -PER_COMMANDER load. Forks register by passing `--fork NAME` to order, which touches -forks/NAME.beat; a beat younger than BEAT_AGE is a live fork. While load needs more forks than are -live, order refuses the main commander's actions and rotate prints FORK NOW, so the only -way forward is forking a slice. +Each open order counts 1, each overdue order 2 more, and every 10 unjudged edits 1. +Forks register with `--fork NAME`; a heartbeat younger than BEAT_AGE is live. """ from __future__ import annotations -import math, os, sys, time +import os, time from dataclasses import dataclass from ..comms import Comms @@ -18,7 +13,6 @@ from .config import CommanderConfig FORKS = f'{CommanderConfig.installed().state}/forks' -PER_COMMANDER = 6 BEAT_AGE = 10 * 60 @@ -50,34 +44,13 @@ def of(cls, comms: Comms, commander: str) -> Load: CommsUnjudgedTool.count(comms, commander), tuple(live_forks())) @property - def judging_load(self) -> int: - """The whole unjudged backlog is one commander's slice, regardless of its length.""" - return PER_COMMANDER if self.unjudged else 0 - - @property - def total(self) -> int: - return self.owed + 2 * self.overdue + self.judging_load - - @property - def needed(self) -> int: - return math.ceil(self.total / PER_COMMANDER) - 1 - - @property - def short(self) -> bool: - """Live forks do not cover the load: the commander must fork now.""" - return len(self.forks) < self.needed + def total(self) -> float: + return self.owed + 2 * self.overdue + self.unjudged / 10 def __str__(self) -> str: return (f'load {self.total:.0f} ({self.owed} open orders, {self.overdue} overdue, ' - f'{self.unjudged} unjudged edits: {self.judging_load} judging load) ' - f'needs {self.needed} live fork(s), ' - f'{len(self.forks)} live {list(self.forks)}; fork a slice now (each fork passes --fork NAME to order)') - - def require(self) -> None: - if self.short: - sys.exit(f'REFUSED, nothing recorded: overloaded: {self}.') + f'{self.unjudged} unjudged edits), {len(self.forks)} live forks {list(self.forks)}') def main(a): - load = Load.of(order.COMMS, order.ME) - print(load if load.short else f'covered: load {load.total:.0f}, forks {list(load.forks)}') + print(Load.of(order.COMMS, order.ME)) diff --git a/src/agent_comms/commander/order.py b/src/agent_comms/commander/order.py index 92f37f8993..de6a972d2e 100644 --- a/src/agent_comms/commander/order.py +++ b/src/agent_comms/commander/order.py @@ -759,21 +759,12 @@ def fire(unit: str, why: str) -> None: print(f'{todo.id}: released from {unit}: {why}') -class LoadedVerb(RuledVerb): - """HARD RULE: overloaded means fork, not more orders from me; a fork is never blocked.""" - - def gate(self) -> None: - super().gate() - if not self.fork: - capacity.Load.of(COMMS, ME).require() - - class OrderVerb(CommanderVerb): """A verb of `order`.""" @dataclass(frozen=True, kw_only=True) -class SendVerb(OrderVerb, LoadedVerb): +class SendVerb(OrderVerb, RuledVerb): help = "record the order in Core and send it" unit: str = option("unit") due: str = option("due", help="HH:MM local time") @@ -916,7 +907,7 @@ def run(self) -> None: @dataclass(frozen=True, kw_only=True) -class SayVerb(OrderVerb, LoadedVerb): +class SayVerb(OrderVerb, RuledVerb): help = "a non-order message; refused if it orders work" units: str = option("units", metavar="UNIT[,UNIT]") body_file: str = option("body_file", metavar="BODYFILE") @@ -926,7 +917,7 @@ def run(self) -> None: @dataclass(frozen=True, kw_only=True) -class ProblemVerb(OrderVerb, LoadedVerb): +class ProblemVerb(OrderVerb, RuledVerb): help = "Core records the problem with its order, or nothing" text: str = option("text") to: str = option("--to") @@ -987,7 +978,7 @@ def run(self) -> None: @dataclass(frozen=True, kw_only=True) -class HuntVerb(OrderVerb, LoadedVerb): +class HuntVerb(OrderVerb, RuledVerb): help = "a hot-path read: clean unless --sent names the order it became" where: str = option("where", metavar="FILE:LINE") why: str = option("--why") diff --git a/src/agent_comms/commander/standing.py b/src/agent_comms/commander/standing.py index b8408a26b7..4232f88329 100644 --- a/src/agent_comms/commander/standing.py +++ b/src/agent_comms/commander/standing.py @@ -17,7 +17,7 @@ from ..threads import Thread from ..todos import FlatTrend, OrderProgress, PullRequests, StrikeTodo, Todo from ..turn_digests import TurnDigestRead, TurnsSince -from . import capacity, judge +from . import judge from .config import CommanderConfig, Repo CONFIG = CommanderConfig.installed() @@ -150,21 +150,6 @@ def enforce(self, comms: Comms, commander: str) -> None: self.assign(comms, commander, comms.registry.require(self.unit)) -@dataclass(frozen=True) -class ForkNow(UndatedStandingRule): - """The load on the asked Comms needs more live forks than there are.""" - - load: capacity.Load - - @classmethod - def violations(cls, comms: Comms, commander: str) -> tuple[ForkNow, ...]: - load = capacity.Load.of(comms, commander) - return (cls(load),) * load.short - - def line(self) -> str: - return f"FORK NOW: {self.load}" - - @dataclass(frozen=True) class Unjudged(ReasonedStandingRule, UndatedStandingRule): limit: ClassVar[int] = 30 diff --git a/tests/test_commander_capacity.py b/tests/test_commander_capacity.py index f20c4124ec..fe0ded1098 100644 --- a/tests/test_commander_capacity.py +++ b/tests/test_commander_capacity.py @@ -1,25 +1,14 @@ -"""The existing capacity gate counts judging as one slice of work.""" +"""Capacity reports the existing load formula and live forks without policy advice.""" import pytest @pytest.mark.parametrize("unjudged", [0, 1, 395, 1_000_000]) -def test_judging_backlog_is_one_slice_used_by_total_and_description(commander_script, unjudged): +def test_capacity_reports_load_and_forks(commander_script, unjudged): script, _ = commander_script - load = script.capacity.Load(24, 14, unjudged, tuple(f"fork-{n}" for n in range(9))) - assert load.judging_load == (script.capacity.PER_COMMANDER if unjudged else 0) - assert load.total == 24 + 2 * 14 + load.judging_load - assert f"{unjudged} unjudged edits: {load.judging_load} judging load" in str(load) - assert load.needed == (9 if unjudged else 8) - assert not load.short - load.require() - - -def test_nine_forks_cover_the_reported_fleet_and_eight_do_not(commander_script): - script, _ = commander_script - forks = tuple(f"fork-{n}" for n in range(9)) - load = script.capacity.Load(24, 14, 395, forks) - assert load.total == 58 and load.needed == 9 - load.require() - with pytest.raises(SystemExit, match="needs 9 live fork"): - script.capacity.Load(24, 14, 395, forks[:-1]).require() + load = script.capacity.Load(24, 14, unjudged, ("judge",)) + assert load.total == 24 + 2 * 14 + unjudged / 10 + assert str(load) == ( + f"load {load.total:.0f} (24 open orders, 14 overdue, {unjudged} unjudged edits), " + "1 live forks ['judge']" + ) diff --git a/tests/test_standing_rules.py b/tests/test_standing_rules.py index e1e5534aa9..beca82731b 100644 --- a/tests/test_standing_rules.py +++ b/tests/test_standing_rules.py @@ -215,34 +215,13 @@ def progress(): assert comms.goals.todos.get("P2").holder == "worker" -def test_fork_now_and_unjudged_follow_the_unjudged_edits(commander, standing): +def test_unjudged_follows_the_unjudged_edits(commander, standing): comms, state, module = standing - assert lines(comms, module.ForkNow) == lines(comms, module.Unjudged) == [] + assert lines(comms, module.Unjudged) == [] create(commander, 'P1') native_judge_edits(commander, module.judge, [dict(id=f'e{n}', at=200., path='a.py', hunks=[dict(old='before', new='after')]) for n in range(70)]) assert lines(comms, module.Unjudged) == ["UNJUDGED 70 edits -> judge or fork a judge slice"] - [fork] = lines(comms, module.ForkNow) - assert fork.startswith("FORK NOW: load 10 (1 open orders, 1 overdue, 70 unjudged edits) needs 1") - - -def test_fork_now_reads_the_load_of_the_bus_it_is_asked_about(commander, standing, tmp_path): - """Three overdue orders on one bus need a fork there; a second bus without orders carries no load.""" - from agent_comms.tool_caller import ThreadCaller - - comms, _, _ = standing - for problem in ("P1", "P2", "P3"): - create(commander, problem) - other = Comms(tmp_path / "other-bus") - other.registry.declare(other.registry.new_thread("lead", frozenset(), str(tmp_path), created_at=1.0)) - - def fork_now(bus): - read = CommsStandingRulesTool().apply(bus, ThreadCaller("lead")) - return [line for line in read.broken if line.startswith("FORK NOW")] - - assert [line.split(") ")[0] for line in fork_now(comms)] == [ - "FORK NOW: load 9 (3 open orders, 3 overdue, 0 unjudged edits"] - assert fork_now(other) == [] def test_rules_due_until_every_drawn_rule_is_answered_and_fresh(commander, monkeypatch):