Parity row31: enforce declared deletion targets on actual delivered commits

plans/commander-native-plan.md

@@ -61,7 +61,7 @@
 | 28 | Measured load and self-report | `capacity.py` unit count | `commander.capacity.Load.of` reports orders, overdue, edits and forks; self-report comparison unverified | `control_plane` |
 | 29 | No handoff of finished, checked work | none; scratch transfer moved P507 after its fix was pushed and checked | Transfer refuses delivered checked work; that owner merges; check is reused | `record_plane` |
 | 30 | A proof is bound to the build it ran on | none; screenshots and logs are bare paths | Build run records runtime and Core/Toad revisions; close requires delivered commit | `record_plane` |
-| 31 | An order states its intent as a deletion target | none; P520 renamed a decision rather than deleting it | Order owns deletion family/site/count; delivery refuses nonzero or renamed decisions | `record_plane` |
+| 31 | An order states its intent as a deletion target | Current order command with a declared FamilyDeletion | CommsTodoCreateTool and CommsOrderDoneTool check the delivered family count | `record_plane` |
 | 32 | Where units' time goes, classified | none; activity.log lacks durations and turn boundaries | `UnitTime`: model, build/check/test/shell, dependency/commander waits and idle; classify large waits | `commander_page`, `control_plane` |
 | 33 | Order progress beside its deadline; the unit never sees a due time | `order.py board`, `overdue`, `due`, `send`; `rotate.py` | Derived `OrderProgress`; merged work draws no strike; first commit 10 min, later progress 15 min STALL; units see no due time | `one_order_owner`, `control_plane` |
 | 34 | Several commanders on one machine and bus, one per project | none; a second commander.json install overwrites the first | Registry commander/project ownership; caller from --from; scoped orders, strikes, load, board, alarms and forks; raw-send guard; duplicate project commander refused | `one_order_owner`, `cutover` |

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

@@ -476,8 +476,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')
 
@@ -729,8 +728,8 @@ def due(self, pid: str, at: str, why: str) -> None:
         moved = deadline(at)
         self.call(CommsChangeProblemTool(problem=pid, operation=AssignTodo(
             todo.revision, todo.assignment, moved, todo.order, todo.goal)))
-        self.on_record(f'{time.strftime("%Y-%m-%d %H:%M")} {todo.holder} ({pid} order {todo.order.id} due '
-                  f'{hhmm(todo.deadline)} moved to {hhmm(moved)}): {why}', 'Deadline',
+        self.on_record(f'{time.strftime("%Y-%m-%d %H:%M")} {todo.holder} ({pid} order {todo.order.id} '
+                  f'deadline changed): {why}', 'Deadline',
                   VerdictAudience.of(todo.held().owner, todo.goal))
         print(f'{pid}: due {hhmm(moved)}')
 

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], ...]: