P310: retain heartbeat finish predictions and compare actual closure

plans/commander-native-plan.md

@@ -57,7 +57,7 @@
 | 24 | Reassigning past two owners needs the commander's trace | scratch `order.py send --reassign --traced` | Transfer reads Todo owner history and requires traced source location | `record_plane` |
 | 25 | Unit wording gets a correction, not a strike | scratch `vibe_watch.py` | `PlainText` for commander verdicts/strikes; unit wording gets correction | `record_plane` |
 | 26 | Alerts carry only decisions | none; every line interrupts | `messages.AlertKind.require_publishable`, `tools.CommsDecisionTool` | `control_plane` |
-| 27 | Momentum prediction per unit, compared with the outcome | none | Prediction record and later outcome on commander page; `CommitMomentum` is a partial candidate | `commander_page` |
+| 27 | Momentum prediction per unit, compared with the outcome | none | Todo retains the latest heartbeat distance forecast and actual closure; OrderProgress shows both, with numeric error only for a measured falling count | `commander_page` |
 | 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` |

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

@@ -465,8 +465,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')
 
@@ -477,12 +476,16 @@ def write_board(self):
                  '| Problem | Owner file | Owner | Order | Due | Check | Progress | Order state '
                  '| Strikes | Owner last acted |',
                  '|---|---|---|---|---|---|---|---|---|---|']
-        listed, read = self.todos(), self.progress()
+        self.todos()  # Reconcile installed merges before reading the board snapshot.
+        snapshot: TodoList = self.call(CommsTodoListTool())
+        listed, read = snapshot.todos, snapshot.progress
         activity = {unit: self.last_activity(unit) for unit in {t.holder for t in listed if t.holder}}
         for t in sorted(listed, key=lambda t: int(t.id[1:]) if t.id[1:].isdigit() else 0):
             lines.append(f'| {t.id}: {t.text.replace("|", "|")} | {t.order.board_cells(t)} | '
                          f'{read[t.id].line if t.id in read else "-"} | {t.order.line} | '
                          f'{t.strike_count} | {activity[t.holder].clock() if t.holder else "-"} |')
+        lines += ['', '## Closed orders: predicted against actual', '',
+                  *(f'- {todo.id}: {read[todo.id].line}' for todo in snapshot.closed)]
         history = self.comms.goals.todos.history_for_commander(self.incarnation)
         lines += ['', '## Withdrawn orders', '',
                   *(f'- {line}' for t in history for line in t.order.withdrawals(t))]
@@ -715,8 +718,8 @@ def due(self, pid: str, at: str, why: str) -> None:
         moved = RecordedEventTime(time.time()).at_clock(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'{todo.deadline.clock()} moved to {moved.clock()}): {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 {moved.clock()}')
 

src/agent_comms/commander_heartbeat.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import time
 from collections.abc import Callable
 from contextlib import AbstractContextManager
 from dataclasses import dataclass, field
@@ -21,10 +22,16 @@ class CommanderReview:
 
     def wake(self, comms: Comms, sender: str, commander: Thread) -> None:
         rules = StandingRule.broken(comms, commander.name)
+        reviewed = comms.goals.todos.review_predictions(
+            commander.incarnation, comms.registry.storage_rule, time.time()
+        )
         found = frozenset(rules)
         if found == self.violations:
             return
-        lines = ["COMMANDER HEARTBEAT: standing-rule violations changed."]
+        lines = [
+            "COMMANDER HEARTBEAT: standing-rule violations changed.",
+            *(f"{todo.id} @{todo.holder}{todo.predicted_finish.line}" for todo in reviewed),
+        ]
         owed = comms.goals.todos.held_for_commander(commander.incarnation)
         units = sorted({todo.held().owner for todo in owed})
         if units:

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