Batch cl-47: P307 P287 P309 (#2119 #2073 #2123)

AGENTS.md

@@ -52,10 +52,8 @@ receipt acknowledgments. Tristan should not have to decode compressed jargon.
   user journey; controlled provider replies are allowed, real UI/transport/state
   are required. A fork check includes immediate opening and its first actual reply.
 - Ship substantial useful checkpoints: implementation owners merge and send
-  Codex the Core/Toad SHAs. Codex alone builds and installs the paired runtime,
-  switches its pointer and operates its restart queue. Other agents may verify
-  supplied builds, but must not run `relink.sh`, publication commands or
-  restart-queue writes, including one-off helpers.
+  the installer the Core/Toad SHAs.
+  Installs run through tools/upgrade_bus.py, by whichever unit the commander assigns, on either machine.
   Deferred CI and final latency targets do not hold those checkpoints.
 - Retain warm rendered bodies across tab returns within the existing resource
   limits. Scroll preparation follows velocity and destination intent, then shrinks

docs/refactor/p79-native-caller.md

@@ -22,4 +22,4 @@ Before/after AST: current source/tests/tools, actual installed Core and metaclas
 
 Local source batch before test trimming: 92 passed, 6 failed. All six fail identically against untouched installed Core 21c25f49d, at wire_log.py:512 with Private bus writer has no durable protocol marker; these are retained baseline fixture failures, assigned to core-retry's existing fixture work. No caller failure is hidden. Ratchet exits 0. `git grep -n 'caller: ToolCaller = NoThreadCaller()' -- src | wc -l` returns 0 on this branch.
 
-Acceptance is a fresh scratch-root native thread on the reviewed installed runtime: actual comms_set_goal records the goal under that thread. Source tests and ratchet are necessary but not installed acceptance. Codex alone builds/installs native and operates the fleet pointer.
+Acceptance is a fresh scratch-root native thread on the reviewed installed runtime: actual comms_set_goal records the goal under that thread. Source tests and ratchet are necessary but not installed acceptance. Installs run through tools/upgrade_bus.py, by whichever unit the commander assigns, on either machine.

src/agent_comms/candidate_maintenance.py

@@ -1,28 +1,22 @@
-"""Best-effort indexing of successful appends after the outer locks release.
-
-Losing a hint or crashing the daemon leaves the optional index unavailable;
-canonical bus and sealed SQL still drive original delivery. Byte positions
-include observations as well as messages, without inventing message sequences.
-"""
+"""Index successful appends after outer locks release, within the owning root's lifetime."""
 
 from __future__ import annotations
 
 import logging
 import sqlite3
 import threading
-from contextlib import ExitStack
 from pathlib import Path
 
 from .bus_route_counts import BusRouteCounts
 from .coordination_errors import CoordinationError
 from .routing import DeliveryMessage
-from .wire_log import CurrentSnapshotOpening, WireLog
 from .wake_candidate_index import FirstBuild, ProjectionUnavailableError, WakeCandidateIndex
+from .wire_log import CurrentSnapshotOpening, WireLog
 
 _LOG = logging.getLogger(__name__)
 _MAX_ACTIVE_ROOTS = 32
 _guard = threading.Lock()
-_pending: dict[Path, tuple[WireLog, int]] = {}
+_pending: dict[Path, CandidateMaintenance] = {}
 
 
 class CandidateMaintenanceFailed(RuntimeError):  # noqa: N818 - named worker failure
@@ -33,72 +27,82 @@ def __init__(self, root: Path) -> None:
         super().__init__(f"Candidate maintenance failed for {root}")
 
 
-def schedule_candidate_catchup(log: WireLog, through_bytes: int) -> None:
-    """Prepare writer indexes, then signal their existing maintenance worker."""
-    if through_bytes <= 0:
-        raise ValueError("candidate notification needs an actual committed byte position")
-    BusRouteCounts(log.path).prepare()
-    root = log.path.parent
-    with _guard:
-        current = _pending.get(root)
-        if current is not None:
-            _pending[root] = (current[0], max(current[1], through_bytes))
-            return
-        if len(_pending) >= _MAX_ACTIVE_ROOTS:
-            raise CandidateMaintenanceFailed(root)
-        _pending[root] = (log, through_bytes)
-        try:
-            threading.Thread(
-                target=_drain_candidate,
-                args=(root,),
-                name="agent-comms-candidate-maintenance",
-                daemon=True,
-            ).start()
-        except RuntimeError as error:
-            _pending.pop(root, None)
-            raise CandidateMaintenanceFailed(root) from error
-
-
-def _retire_candidate(root: Path) -> None:
-    with _guard:
-        _pending.pop(root, None)
+class CandidateMaintenance(threading.Thread):
+    """One root's coalesced cursor and actual worker; retirement joins this owner."""
 
+    def __init__(self, log: WireLog, through_bytes: int) -> None:
+        super().__init__(name="agent-comms-candidate-maintenance", daemon=True)
+        self.log = log
+        self.through_bytes = through_bytes
 
-def _drain_candidate(root: Path) -> None:
-    with ExitStack() as cleanup:
-        cleanup.callback(_retire_candidate, root)
+    def run(self) -> None:
+        root = self.log.path.parent
         try:
-            with _guard:
-                log, _bytes = _pending[root]
-            # Source ownership is checked off the producer path. Maintenance cannot
-            # create the protocol marker or authorize delivery.
-            with log.checked_read() as source:
+            with self.log.checked_read() as source:
                 metadata = source.marker
             root_id = metadata.root_id
-            index = WakeCandidateIndex(log)
-            routes = BusRouteCounts(log.path)
+            index = WakeCandidateIndex(self.log)
+            routes = BusRouteCounts(self.log.path)
             while True:
                 with _guard:
-                    target = _pending[root][1]
+                    target = self.through_bytes
                 hint = index.notify_committed_append(root_id=root_id, through_bytes=target)
-                result = index.catch_up_committed_append(
-                    hint, start=FirstBuild()
-                )
-                with log.opened_wire_snapshot(opening=CurrentSnapshotOpening()) as snapshot:
+                result = index.catch_up_committed_append(hint, start=FirstBuild())
+                with self.log.opened_wire_snapshot(opening=CurrentSnapshotOpening()) as snapshot:
                     route_result = routes.sync(
                         snapshot,
-                        lambda record: DeliveryMessage.from_wire(record, root_id=snapshot.metadata.root_id),
+                        lambda record: DeliveryMessage.from_wire(
+                            record, root_id=snapshot.metadata.root_id
+                        ),
                     )
                 if result.caught_up and route_result.complete:
                     with _guard:
-                        if _pending[root][1] <= result.checkpoint_bytes:
-                            # Disarm cleanup before a successor can own this root.
-                            cleanup.pop_all()
+                        if self.through_bytes <= result.checkpoint_bytes:
                             _pending.pop(root)
                             return
         except (
-            OSError, ValueError, sqlite3.Error, CoordinationError, ProjectionUnavailableError,
+            OSError,
+            ValueError,
+            sqlite3.Error,
+            CoordinationError,
+            ProjectionUnavailableError,
         ) as error:
             failure = CandidateMaintenanceFailed(root)
             _LOG.exception("%s", failure)
             raise failure from error
+        finally:
+            with _guard:
+                if root in _pending and _pending[root] is self:
+                    _pending.pop(root)
+
+
+def schedule_candidate_catchup(log: WireLog, through_bytes: int) -> None:
+    """Prepare writer indexes, then coalesce into the root's actual worker."""
+    if through_bytes <= 0:
+        raise ValueError("candidate notification needs an actual committed byte position")
+    BusRouteCounts(log.path).prepare()
+    root = log.path.parent
+    with _guard:
+        if root in _pending:
+            worker = _pending[root]
+            worker.through_bytes = max(worker.through_bytes, through_bytes)
+            return
+        if len(_pending) >= _MAX_ACTIVE_ROOTS:
+            raise CandidateMaintenanceFailed(root)
+        worker = CandidateMaintenance(log, through_bytes)
+        _pending[root] = worker
+        try:
+            worker.start()
+        except RuntimeError as error:
+            _pending.pop(root)
+            raise CandidateMaintenanceFailed(root) from error
+
+
+def drain_candidate_catchup(root: Path) -> None:
+    """A stopped root cannot be renamed or removed while its worker still writes."""
+    while True:
+        with _guard:
+            if root not in _pending:
+                return
+            worker = _pending[root]
+        worker.join()

src/agent_comms/check_root.py

@@ -12,6 +12,7 @@
 from typing import ClassVar
 
 from .attempt_recovery import RecoveryMonitorCapability
+from .candidate_maintenance import drain_candidate_catchup
 from .child_process import NoProcess
 from .cli_result import CliResult
 from .comms import Comms
@@ -74,6 +75,7 @@ def open(self) -> Iterator[Comms]:
                 ):
                     copied.owners.stop(thread.name)
         finally:
+            drain_candidate_catchup(root)
             RecoveryMonitorCapability.retire_check_root(root)
 
     def _populate(self, copied: Comms) -> None:

src/agent_comms/commander/order.py

@@ -44,7 +44,7 @@
 from ..goals import OrderCheck, ProblemGoalOrder
 from ..message_reference import MessageReference
 from ..native_tool_call import SelectedToolDenied
-from ..plan_phases import OrderPhase
+from ..plan_phases import OrderPhase, PlanPhase
 from ..progress_claims import MergedClaim, WorkingClaim
 from ..recorded_time import RecordedEventTime
 from ..repo import Repo
@@ -169,6 +169,7 @@ class OrderFile:
     order: PendingProblemOrder
     defect: str
     why_stupid: str
+    unplanned_reason: str = ""
     problems: tuple[str, ...] = ()
     slop: tuple[Slop, ...] = ()
 
@@ -181,7 +182,8 @@ def read(cls, text: str):
 
     @property
     def prose(self) -> str:
-        return f'{self.defect}\n{self.why_stupid}'
+        reason = f'\nUnplanned reason: {self.unplanned_reason}' if self.unplanned_reason else ''
+        return f'{self.defect}\n{self.why_stupid}{reason}'
 
     def problems_found(self, clean: bool) -> list[str]:
         problems = [f'{name}: {error}; fix: state the defect plainly without wording rules on code'
@@ -325,6 +327,20 @@ def send(self, unit, due, document: OrderFile, clean=False, reassign=None, new=N
             problems.append('due: fix: supply HH:MM local time (00:00 through 23:59)')
         group = [new[0]] if new else list(document.problems)
         listed = {t.id: t for t in self.todos()}
+        history = self.comms.goals.todos.history_for_commander(self.incarnation)
+        problems.extend(order.phase.order_refusals(history, document.unplanned_reason))
+        held = tuple(todo for todo in listed.values() if todo.holder)
+        busy = tuple(todo for plan in PlanPhase.members_with(PlanPhase)
+                     for todo in plan.serving(held) if todo.holder == unit and todo.id not in group)
+        if busy:
+            snapshot = self.comms.registry.snapshot()
+            occupied = {todo.holder for todo in held} | {self.commander} | self.parked
+            idle = sorted(thread.name for thread in snapshot.threads.values()
+                          if thread.name not in occupied and thread.role.executor(thread)
+                          and snapshot.statuses[thread.name].active)
+            problems.append(f'{unit} already holds plan orders '
+                            f'{", ".join(todo.id for todo in busy)}; '
+                            f'running units holding no order: {", ".join(idle) or "none"}')
         if not group or any(not re.fullmatch(r'P[1-9]\d*', pid) for pid in group):
             problems.append('problems: fix: supply an array of P<n> IDs, or use order problem')
         if new and document.problems:
@@ -480,8 +496,7 @@ def problem(self, order_file, unit, due, where, disk=False,
     def close(self, pid, works, artifact=''):
         """Core owns closing: the installed evidence, the artifact and the no-open-order rule live on the Todo."""
         evidence = artifact or self.comms.goals.todos.get(pid).order.closing_artifact()
-        ThreadCaller(self.commander).invoke(CommsProblemCloseTool(problem=pid,
-            works=works, artifact=evidence), self.comms)
+        self.call(CommsProblemCloseTool(problem=pid, works=works, artifact=evidence))
         self.write_board()
         print(f'{pid} closed')
 
@@ -802,6 +817,23 @@ def parse(cls, argv: list[str], description: str) -> CommanderVerb:
         return cls.member_named(args.command).from_arguments(args)
 
 
+class StaleOrderGate(CommanderVerb):
+    """All directive verbs read the configured commander's same retained board."""
+
+    def gate(self, ctx: Commander) -> None:
+        super().gate(ctx)
+        stale = tuple(todo for todo in ctx.held_orders() if todo.overdue(time.time() - 30 * 60))
+        if stale:
+            lines = []
+            for todo in stale:
+                due = time.localtime(todo.deadline.recorded_at("An order deadline"))
+                lines.append(f'{todo.id}: holder {todo.holder}, deadline '
+                             f'{time.strftime("%Y-%m-%d %H:%M", due)}')
+            sys.exit('REFUSED, nothing recorded: orders more than 30 minutes overdue:\n  '
+                     + '\n  '.join(lines)
+                     + '\nMove their deadlines or withdraw them before sending more work.')
+
+
 class RuledVerb(CommanderVerb):
     """HARD RULE: no commander action, mine or a fork's, on rules not re-read in 20 min."""
 
@@ -855,7 +887,7 @@ class OrderVerb(CommanderVerb):
 
 
 @dataclass(frozen=True, kw_only=True)
-class SendVerb(OrderVerb, DirectiveVerb):
+class SendVerb(OrderVerb, DirectiveVerb, StaleOrderGate):
     help = "record the order in Core and send it"
     unit: str = option("unit")
     due: str = option("due", help="HH:MM local time")
@@ -874,7 +906,7 @@ def run(self, ctx: Commander) -> None:
 
 
 @dataclass(frozen=True, kw_only=True)
-class StrikeVerb(OrderVerb, DirectiveVerb):
+class StrikeVerb(OrderVerb, DirectiveVerb, StaleOrderGate):
     help = "a send-back, a missed deadline or a false claim"
     unit: str = option("unit")
     why: str = option("--why")
@@ -971,9 +1003,11 @@ class PullRequestVerb(OrderVerb, declared_name="pr"):
 
     def run(self, ctx: Commander) -> None:
         todo = ctx.comms.goals.todos.get(self.pid)
-        checkout = ctx.comms.registry.storage_rule.order_checkout(todo.holder, todo.id)
+        checkout = Repo.of(Path.cwd())
+        if checkout.github.lower() != todo.repo.lower():
+            sys.exit(f'Refused: current checkout belongs to {checkout.github}, not {todo.repo}.')
         print(todo.order.open_pull_request(
-            todo, checkout, self.title, Path(self.body_file).read_text()))
+            todo, checkout.checkout, self.title, Path(self.body_file).read_text()))
 
 
 @dataclass(frozen=True, kw_only=True)
@@ -999,7 +1033,7 @@ def run(self, ctx: Commander) -> None:
 
 
 @dataclass(frozen=True, kw_only=True)
-class SayVerb(OrderVerb, DirectiveVerb):
+class SayVerb(OrderVerb, DirectiveVerb, StaleOrderGate):
     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")
@@ -1009,7 +1043,7 @@ def run(self, ctx: Commander) -> None:
 
 
 @dataclass(frozen=True, kw_only=True)
-class ProblemVerb(OrderVerb, DirectiveVerb):
+class ProblemVerb(OrderVerb, DirectiveVerb, StaleOrderGate):
     help = "Core records the problem with its order, or nothing"
     order: str = option("order", metavar="ORDER.json")
     to: str = option("--to")

src/agent_comms/commander_parity/__init__.py

@@ -3,7 +3,7 @@
 from __future__ import annotations
 
 from abc import ABC, abstractmethod
-from collections.abc import Iterable
+from collections.abc import Callable, Iterable
 from dataclasses import dataclass
 from pathlib import Path
 from typing import TYPE_CHECKING, ClassVar
@@ -43,6 +43,7 @@ class ParityRow(
     current: ClassVar[str]
     native: ClassVar[str]
     phases: ClassVar[tuple[type[PlanPhase], ...]]
+    comparison: ClassVar[Callable[[], ParityComparison]] = UnverifiedParity
 
     @classmethod
     def compare(cls, todos: Iterable[Todo]) -> ParityComparison:
@@ -51,7 +52,7 @@ def compare(cls, todos: Iterable[Todo]) -> ParityComparison:
     @classmethod
     def verify(cls) -> ParityComparison:
         """Run this declared workflow now; a run is not durable cutover authority."""
-        return UnverifiedParity()
+        return cls.comparison()
 
     @classmethod
     def rows(cls) -> tuple[type[ParityRow], ...]: