Integrate #1972 #1976

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,16 @@ def recover(
         path: Path,
         saved: PrefixWitness,
         info: os.stat_result,
-    ) -> PrefixWitness | None: ...
+    ) -> 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: ...
+
+    @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.")
@@ -90,7 +99,8 @@ def capture(cls, witness: PrefixWitness, db_path: Path) -> FinalSeal:
 
     def check_final(self, saved: PrefixWitness, db_path: Path) -> None:
         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,
@@ -100,8 +110,17 @@ def recover(
         path: Path,
         saved: PrefixWitness,
         info: os.stat_result,
-    ) -> None:
+    ) -> PrefixWitness:
         self.check_final(saved, path)
+        if saved.revision == FileRevision.from_stat(info):
+            return saved
+        return super().recover(bus, marker, db, path, saved, info)
+
+    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():
@@ -129,15 +148,21 @@ 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 != FileRevision.from_stat(info)
+            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
+
+    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 (

src/agent_comms/commander/capacity.py

@@ -1,23 +1,18 @@
-"""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, 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
-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
 from . import order
 from .config import CommanderConfig
 
 FORKS = f'{CommanderConfig.installed().state}/forks'
-PER_COMMANDER = 6
 BEAT_AGE = 10 * 60
 
 
@@ -52,25 +47,10 @@ def of(cls, comms: Comms, commander: str) -> Load:
     def total(self) -> float:
         return self.owed + 2 * self.overdue + self.unjudged / 10
 
-    @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 __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'{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))

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")

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

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."""