Batch round 0805: nine ready heads, merged and tested once

.github/workflows/ci.yml

@@ -104,14 +104,14 @@ jobs:
       - name: Run tests with coverage (85% enforced on Linux)
         if: matrix.os == 'ubuntu-latest'
         run: |
-          pytest --cov=agent_comms --cov-report=xml --cov-report=term-missing -v
+          pytest -n auto --maxprocesses=8 --dist=worksteal --cov=agent_comms --cov-report=xml --cov-report=term-missing -v
 
       # Sealed /var/tmp, /proc and SO_PEERCRED fixtures run only on Linux.
       # Keep macOS test execution, but do not enforce Linux coverage there.
       - name: Run tests (coverage reported, Linux-only fixtures skipped)
         if: matrix.os != 'ubuntu-latest'
         run: |
-          pytest --cov=agent_comms --cov-fail-under=0 --cov-report=xml --cov-report=term-missing -v
+          pytest -n auto --maxprocesses=8 --dist=worksteal --cov=agent_comms --cov-fail-under=0 --cov-report=xml --cov-report=term-missing -v
 
   build:
     runs-on: ubuntu-latest

README.md

@@ -112,9 +112,9 @@ It does not publish the runtime or change the live installation. Pass `--home DI
 to retain a private test build for inspection. The pytest basetemp is disposable,
 named for the checkout's HEAD, and the test run is limited to 300 seconds.
 
-Pytest uses separate `pytest-xdist` worker processes (automatic CPU detection,
-capped at eight) with work stealing and combined coverage. Use `pytest -n 0`
-for serial debugging or `pytest -n 4` to choose a worker count.
+Pytest runs serially by default, so a run of a few touched tests starts no
+worker pool. The full suite in CI passes `-n auto --maxprocesses=8
+--dist=worksteal` itself; do the same locally for a full run.
 
 ## Releasing
 

pyproject.toml

@@ -101,7 +101,6 @@ pythonpath = [".", "src", "tools"]
 python_files = ["test_*.py"]
 python_classes = ["Test*"]
 python_functions = ["test_*"]
-addopts = "-n auto --maxprocesses=8 --dist=worksteal"
 asyncio_mode = "auto"
 # A hung test fails by name instead of holding the whole job.
 timeout = 60

src/agent_comms/acp_failure.py

@@ -173,7 +173,6 @@ def summary(self) -> str:
 class ACPFailure(DeliveryFailure):
     code: int | None
     detail: str
-    classification_priority: ClassVar[int] = 0
     input_state: type[InputAttempt] | None = None
     diagnostics: tuple[PiDiagnostic, ...] = ()
     # Public failed-outcome projection of the turn-owned immutable result.
@@ -224,11 +223,7 @@ def from_detail(cls, code, detail, *, diagnostics=()):
         detail = detail or "ACP request failed"
         owner = next(
             member
-            for member in sorted(
-                cls.members_with(ACPFailure),
-                key=lambda member: member.classification_priority,
-                reverse=True,
-            )
+            for member in cls.members_with(ACPFailure)
             if member.matches(code, detail, diagnostics)
         )
         return owner(code, detail, diagnostics=diagnostics)
@@ -346,20 +341,33 @@ def raise_if_failed(self) -> NoReturn:
         raise NativeTurnError(self.request)
 
 
-class RequestACPFailure(ACPFailure):
+class ProviderConnectionFailure(ACPFailure):
     @classmethod
-    def matches(
-        cls, code: int | None, detail: str, diagnostics: tuple[PiDiagnostic, ...] = ()
-    ) -> bool:
-        return True
+    def matches(cls, code, detail, diagnostics=()) -> bool:
+        return any(diagnostic.provider_connection_failure for diagnostic in diagnostics)
 
     @property
     def title(self) -> str:
-        return "Request failed"
+        return "Provider connection failed"
+
+    @property
+    def description(self) -> str:
+        known = "\n".join(
+            diagnostic.description
+            for diagnostic in self.diagnostics
+            if diagnostic.provider_connection_failure
+        )
+        return "\n".join(part for part in (self.detail, known) if part)
+
+    @property
+    def next_step(self) -> str:
+        return (
+            "The provider connection failed. Inspect the response and input delivery "
+            "before choosing whether to send a new message."
+        )
 
 
 class ProviderQuotaFailure(ACPFailure):
-    classification_priority = 100
     display_pattern = re.compile(
         r"usage limit|quota|insufficient credits|credit balance|rate.limit|too many requests",
         re.I,
@@ -383,32 +391,16 @@ def next_step(self) -> str:
         )
 
 
-class ProviderConnectionFailure(ACPFailure):
-    classification_priority = 150
-
+class RequestACPFailure(ACPFailure):
     @classmethod
-    def matches(cls, code, detail, diagnostics=()) -> bool:
-        return any(diagnostic.provider_connection_failure for diagnostic in diagnostics)
+    def matches(
+        cls, code: int | None, detail: str, diagnostics: tuple[PiDiagnostic, ...] = ()
+    ) -> bool:
+        return True
 
     @property
     def title(self) -> str:
-        return "Provider connection failed"
-
-    @property
-    def description(self) -> str:
-        known = "\n".join(
-            diagnostic.description
-            for diagnostic in self.diagnostics
-            if diagnostic.provider_connection_failure
-        )
-        return "\n".join(part for part in (self.detail, known) if part)
-
-    @property
-    def next_step(self) -> str:
-        return (
-            "The provider connection failed. Inspect the response and input delivery "
-            "before choosing whether to send a new message."
-        )
+        return "Request failed"
 
 
 class ErrorValue(DeclaredFamily, JsonShapeFamily, affix="ErrorValue"):

src/agent_comms/bus_tables.py

@@ -3,16 +3,19 @@
 from __future__ import annotations
 
 import sqlite3
-from collections.abc import Iterator
+from collections.abc import Iterable, Iterator
+from abc import ABC, abstractmethod
 from contextlib import closing
-from dataclasses import Field, dataclass, fields, make_dataclass
+from dataclasses import Field, dataclass, field, fields, make_dataclass
 from functools import cache
 from inspect import get_annotations
+from itertools import accumulate
 from typing import TYPE_CHECKING, Annotated, ClassVar, Self, get_origin, get_type_hints
 
 from agent_comms import messages
 from agent_comms.coordination_errors import IdentityConflict
 from agent_comms.envelope_claim_transitions import (
+    ClaimOwner,
     ClaimProjection,
     ClaimTransition,
     ClaimTransitionError,
@@ -54,9 +57,9 @@ class BusProjectionTables(ProducedRow, SharedFileTables, DerivedTables):
     def create_schema(cls, db: sqlite3.Connection) -> None:
         """Create/rebuild rows from history using the append projection owner."""
         super().create_schema(db)
-        with closing(BusRow.stream(db, order_by=("seq",))) as headers:
-            for header in headers:
-                header.kind.read_from(db, header.seq).project_into(db)
+        holders = ReplayedHolders()
+        for record in BusCapability.history(db):
+            record.project_into(db, holders)
 
 
 class BusSchema(ComposedSchemas, BusTables, ProjectionProducerTables, BusProjectionTables):
@@ -357,17 +360,20 @@ def _project(
         })
 
     @classmethod
-    def project(cls, db: sqlite3.Connection, transition: ClaimTransition) -> ClaimProjection:
+    def project(
+        cls, db: sqlite3.Connection, transition: ClaimTransition, holders: ClaimHolders,
+    ) -> ClaimProjection:
         """Validate before the original is inserted; the caller retains the writer."""
         try:
             transition.require_new_operation(lambda operation: ClaimOperations.used(db, operation))
-            return transition.project(cls.claims_for(db, transition.resources, transition.seq - 1))
+            return transition.project(holders.before(db, transition))
         except (ClaimTransitionError, IdentityConflict) as error:
             raise RelationViolationError("Committed claim transition is inconsistent.") from error
 
     @classmethod
     def apply(
         cls, db: sqlite3.Connection, transition: ClaimTransition, projected: ClaimProjection,
+        holders: ClaimHolders,
     ) -> None:
         removed = transition.resources + transition.expired
         marks = ",".join("?" for _ in removed)
@@ -376,6 +382,47 @@ def apply(
             cls(owner.resource, owner.seq).insert(db)
         for operation_id in transition.operation_ids:
             ClaimOperations(operation_id, transition.seq).insert(db)
+        holders.advance(removed, projected)
+
+
+class ClaimHolders(ABC):
+    """The current holders a claim transition projects from, before it is applied."""
+
+    @abstractmethod
+    def before(self, db: sqlite3.Connection, transition: ClaimTransition) -> ClaimProjection: ...
+
+    def advance(self, removed: tuple[str, ...], projected: ClaimProjection) -> None:
+        """Held rows already record the applied transition."""
+
+
+class StoredHolders(ClaimHolders):
+    """One append reads the holders of its resources from their table."""
+
+    def before(self, db: sqlite3.Connection, transition: ClaimTransition) -> ClaimProjection:
+        return HeldResources.claims_for(db, transition.resources, transition.seq - 1)
+
+
+@dataclass
+class ReplayedHolders(ClaimHolders):
+    """An in-order rebuild carries the holders it already projected; it never rereads them.
+
+    Each owner was validated in the projection that produced it; only the subset a
+    transition projects from is validated again, as one ``ClaimProjection``.
+    """
+
+    last_seq: int = 0
+    owners: dict[str, ClaimOwner] = field(default_factory=dict)
+
+    def before(self, db: sqlite3.Connection, transition: ClaimTransition) -> ClaimProjection:
+        return ClaimProjection(self.last_seq, {
+            resource: self.owners[resource]
+            for resource in transition.resources if resource in self.owners})
+
+    def advance(self, removed: tuple[str, ...], projected: ClaimProjection) -> None:
+        for resource in removed:
+            self.owners.pop(resource, None)
+        self.owners.update(projected)
+        self.last_seq = projected.last_seq
 
 
 @dataclass(frozen=True)
@@ -438,18 +485,42 @@ def rows(self) -> tuple[CapabilityRow, ...]:
     @classmethod
     def read_from(cls, db: sqlite3.Connection, seq: int) -> Self:
         """Reconstruct this declared value from its own capability tables."""
-        values = {}
+        rows = []
         for table in cls.tables():
-            rows = table.select(db, seq=seq)
-            if not rows:
+            stored = table.select(db, seq=seq)
+            if not stored:
                 raise RelationViolationError(f"Bus row {seq} lacks {table.declared_name}")
-            values.update({item.name: getattr(rows[0], item.name) for item in fields(table)})
-        return FieldCodec.construct(cls, values)
+            rows.append(stored[0])
+        return cls.assembled(rows)
+
+    @classmethod
+    def assembled(cls, rows: Iterable[CapabilityRow]) -> Self:
+        """This declared value from one stored row of each of its capability tables."""
+        return FieldCodec.construct(cls, {
+            item.name: getattr(row, item.name) for row in rows for item in fields(row)})
+
+    @staticmethod
+    def history(db: sqlite3.Connection) -> Iterator[BusCapability]:
+        """Every record in sequence order from one joined cursor, each stored row decoded once."""
+        tables = TypedTable.members_with(CapabilityRow)
+        offsets = dict(zip(tables, accumulate(
+            (len(BusRow.columns()), *(len(table.columns()) for table in tables)))))
+        header = BusRow.qualified("seq")
+        with closing(db.execute(
+            f"SELECT {', '.join(row.qualified(*row.column_projection()) for row in (BusRow, *tables))} "
+            f'FROM "{BusRow.declared_name}" '
+            + " ".join(f'LEFT JOIN "{table.declared_name}" ON {table.qualified("seq")} = {header}'
+                       for table in tables)
+            + f" ORDER BY {header}"
+        )) as cursor:
+            for stored in cursor:
+                kind = BusRow.decode_at(stored, 0).kind
+                yield kind.assembled(table.decode_at(stored, offsets[table]) for table in kind.tables())
 
     def projections(self) -> Iterator[TypedTable]:
         return iter(())
 
-    def project_into(self, db: sqlite3.Connection) -> None:
+    def project_into(self, db: sqlite3.Connection, holders: ClaimHolders) -> None:
         for row in self.projections():
             row.insert(db)
 

src/agent_comms/check_root.py

@@ -29,6 +29,7 @@
 from .session_file import SavedSessionFile
 from .threads import Thread
 from .transcript_routes import TranscriptRoutes
+from .turn_lease import NoTurnRecord
 
 
 @dataclass(frozen=True)
@@ -105,7 +106,8 @@ def _populate(self, copied: Comms) -> None:
             project.mkdir(parents=True, mode=0o700)
             session = self._session(original, copied.root, slot, project)
             declarations[name] = replace(
-                original.without_turn(),
+                original,
+                turn_record=NoTurnRecord(),
                 worktree=str(project),
                 parent=original.parent.kept_in(selected),
                 task=NoThreadText(),
@@ -116,7 +118,6 @@ def _populate(self, copied: Comms) -> None:
                 project_resume_due=False,
                 storage_roots=StorageRoots(copied.root / "scratch" / slot),
                 turn_generation=0,
-                last_finished_turn_id=None,
             )
         with copied.registry.store.editing() as edit:
             edit.document.restore_stopped(snapshot, names)

src/agent_comms/commander/rule_file.py

@@ -115,21 +115,14 @@ def drift_tells(cls) -> tuple[str, ...]:
         """The reminder hook's one-line tells, one declared by each packaged rule."""
         return tuple(rule.drift for rule in cls.packaged())
 
-    @abstractmethod
-    def review_rules(self) -> tuple[str, ...]: ...
-
 
 class FeedbackRuleFile(RuleFile):
-
-    def review_rules(self) -> tuple[str, ...]:
-        return (self.summary,)
+    """A rule that supplies feedback in the commander's instructions."""
 
 
 class HardRuleFile(FeedbackRuleFile, declared_name="hard-rule"):
     """A feedback rule that is never waived; refining FeedbackRuleFile puts it first."""
 
 
 class ReferenceRuleFile(RuleFile):
-
-    def review_rules(self) -> tuple[str, ...]:
-        return ()
+    """Supporting instructions read alongside the feedback rules."""