P439 derive one sibling UnitState per unit

plans/commander-native-plan.md

@@ -157,7 +157,7 @@ Commanding is a high-tempo RTS. The fleet loses information because Core hands t
 The fix is one derived model with four renderers. **Core is the engine, Toad is the screen, the commander's context is the HUD feed, and the connector serves claude-ai the same frames.** Nothing computes fleet state twice.
 
 **Engine: `CommanderHud` in `ui_model`, derived on read, holding no state but one cursor.**
-- **`UnitState` family.** The rotation tick's classification moves out of SKILL prose into Core: `Blocked` (goal execution blocked while holding an order), `Detached` (not receiving), `AwaitingCommander` (a question, alert or handoff with no commander answer, from `Message.answers`), `Stuck` (no tool call while holding an order), `Spinning` (send-backs or reruns with no new commit), `StalledOnNothing` (turn ended holding an order with nothing pending), `DeliveredUnclosed` (merged or installed, order open), `Moving` (fresh commit on the order's surface), `Idle` (no order). Each member derives itself from turn digests, git, goal execution and messages; owns its card line, its ping and its default action. Urgency order is the MRO, never a rank field.
+- **`UnitState` family.** The rotation tick's classification moves out of SKILL prose into Core: `Unreadable` (a checkout or other unit source cannot be read; other units still get their cards), `Blocked` (goal execution blocked while holding an order), `Detached` (not receiving), `AwaitingCommander` (a question, alert or handoff with no commander answer, from `Message.answers`), `Stuck` (no tool call while holding an order), `Spinning` (send-backs or reruns with no new commit), `StalledOnNothing` (turn ended holding an order with nothing pending), `DeliveredUnclosed` (merged or installed, order open), `Moving` (fresh commit on the order's surface), `Idle` (no order). Each member derives itself from turn digests, git, goal execution and messages; owns its card line, its ping and its default action. States are direct `UnitState` members, sharing behavior through capability mixins rather than claiming to be other states. Urgency is their declaration order in the family's registry, never a rank field.
 - **Pings are decisions, each once.** A unit entering an urgent state, a review timer up, a check failed, a merge, an install, a run seen working, an owner message. Keyed by unit, state and order revision, so nothing re-fires. The `StandingRule` family is the ping source; `HolderCannotRun` (with `Detached` and `Blocked` as members) and `UnitAskUnanswered` join it.
 - **Frames.** `CommanderHud.frame()` is a snapshot; `frame.since(cursor)` is the delta: only changed cards and new pings, like a netcode delta. The one stored fact is each reader's cursor.
 - **Kill feed:** `comms_fleet_story` plus merges and installs, newest first. **Economy bar:** free disk against floor, merge-queue depth, units busy, idle and blocked, unjudged edits, model budget where the provider reports it. **Scoreboard:** per unit, delivered and installed orders, strikes, and self-inflicted wait minutes (`UnitTime`, row 32).

src/agent_comms/active_route.py

@@ -147,7 +147,7 @@ def wire_log(self) -> WireLog:
     def selects_its_root(self) -> bool:
         """Whether the root still exists and carries the marker this route was published for."""
         return self.root.is_dir() and (
-            self.wire_log().read_metadata_unlocked().wire_root_id == self.wire_root_id)
+            self.wire_log().read_metadata_unlocked().root_id == self.wire_root_id)
 
     def observe_root(self) -> Path:
         # The writer atomically replaces this marker. Reuse its owner/mode,

src/agent_comms/commander/config.py

@@ -9,10 +9,11 @@
 from dataclasses import dataclass
 from importlib.resources import files
 from pathlib import Path
-from typing import TYPE_CHECKING, Annotated, Self
+from typing import TYPE_CHECKING, Annotated, ClassVar, Self
 
 from pydantic import BaseModel, ConfigDict, ValidationInfo, field_validator
 
+from agent_comms.locked_store import DurableFile
 from agent_comms.phase_evidence import MainTree, PlanReading
 from agent_comms.plan_phases import PlanPhase
 
@@ -48,7 +49,8 @@ def require_commander(thread: Thread) -> None:
             )
 
 
-class CommanderConfig(CommanderRole, BaseModel):
+class CommanderConfig(CommanderRole, BaseModel, DurableFile):
+    filename: ClassVar[str] = "commander.json"
     model_config = ConfigDict(frozen=True)
 
     commander: str
@@ -62,7 +64,7 @@ class CommanderConfig(CommanderRole, BaseModel):
 
     @staticmethod
     def path(home: Path | None = None) -> Path:
-        return (home or Path.home()) / '.config/agent-comms/commander.json'
+        return (home or Path.home()) / '.config/agent-comms' / CommanderConfig.filename
 
     @staticmethod
     def packaged_rules() -> Path:
@@ -91,14 +93,10 @@ def selected_root(cls, root: Path, info: ValidationInfo) -> Path:
     def read(cls, path: Path | str) -> Self:
         return cls.model_validate_json(Path(path).read_text())
 
-    @staticmethod
-    def root_path(root: Path) -> Path:
-        return root / 'commander.json'
-
     @classmethod
     def read_root(cls, root: Path | str) -> Self:
         root = Path(root).resolve()
-        return cls.model_validate_json(cls.root_path(root).read_text(), context=root)
+        return cls.model_validate_json(cls.path_in(root).read_text(), context=root)
 
     @classmethod
     def on_machine(cls, commander: str, repos: tuple[Repo, ...], root: Path, home: Path,
@@ -143,7 +141,7 @@ def write(self, home: Path) -> Path:
         self.require_commander(Comms(self.root).registry.require(self.commander))
         path = self.path(home)
         path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
-        record = self.root_path(self.root)
+        record = self.path_in(self.root)
         record.write_text(self.model_dump_json(indent=1) + '\n')
         path.unlink(missing_ok=True)
         path.symlink_to(record)
@@ -160,7 +158,7 @@ class CommanderRecord(ABC):
     @boundary_decoder
     def read(root: Path) -> CommanderRecord:
         root = root.resolve()
-        if CommanderConfig.root_path(root).is_file():
+        if CommanderConfig.path_in(root).is_file():
             return RecordedCommander(CommanderConfig.read_root(root))
         return NoCommanderRecord()
 

src/agent_comms/commander/order.py

@@ -839,7 +839,7 @@ def answer_references(self, ctx: Commander) -> tuple[MessageReference, ...]:
 
     def gate(self, ctx: Commander) -> None:
         named = self.named(ctx)
-        owed = tuple(rule for rule in OwedAnswer.broken(ctx.comms, self.who(ctx))
+        owed = tuple(rule for rule in OwedAnswer.overdue(ctx.comms, self.who(ctx))
                      if not rule.answered_by(named))
         if owed:
             lines = '\n'.join(f'{rule.line()}: {rule.message.body[:80]}' for rule in owed)
@@ -1249,7 +1249,8 @@ class NextHuntVerb(OrderVerb, declared_name="next-hunt"):
     help = "print the next place to hunt"
 
     def run(self, ctx: Commander) -> None:
-        for path in HuntDue.next_hunt(ctx.comms, ctx.commander):
+        for path in HuntDue.next_hunt(ctx.comms, ctx.commander,
+                                      HuntDue.todos(ctx.comms, ctx.commander)):
             print(path)
 
 

src/agent_comms/commander/questions.py

@@ -0,0 +1,112 @@
+"""Message reads shared by unit observations and commander answer rules."""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+from contextlib import closing
+from dataclasses import dataclass
+from typing import TYPE_CHECKING, ClassVar
+
+from agent_comms.bus_page_index import BusPageIndex
+from agent_comms.message_page import MessageReadRequest
+from agent_comms.message_reference import MessageReference
+from agent_comms.messages import Message, NeedsAnswerKind
+from agent_comms.read_basis import RepliesDisplayScope
+from agent_comms.wire_log import WireSnapshot
+
+if TYPE_CHECKING:
+    from agent_comms.comms import Comms
+    from agent_comms.todos import Todo
+
+
+@dataclass(frozen=True, slots=True)
+class UnansweredUnitAsks(MessageReadRequest[tuple[Message, ...]]):
+    scope: RepliesDisplayScope
+    since: float
+    before: float
+    empty: ClassVar[tuple[Message, ...]] = ()
+
+    @classmethod
+    def for_orders(
+        cls, comms: Comms, commander: str, todos: Sequence[Todo], before: float
+    ) -> tuple[Message, ...]:
+        if not todos:
+            return ()
+        snapshot = comms.registry.snapshot()
+        peers = frozenset(name for todo in todos for name in snapshot.names_for(todo.held().owner))
+        since = min(
+            order.at.recorded_at("A held order") for todo in todos for order in todo.order_history
+        )
+        questions = comms.bus.log.read_page(
+            cls(
+                RepliesDisplayScope(snapshot.names_for(commander), peers),
+                since,
+                before,
+            )
+        )
+        return tuple(message for message in questions if issubclass(message.type, NeedsAnswerKind))
+
+    def indexed(self, index: BusPageIndex, source: WireSnapshot) -> tuple[Message, ...]:
+        answered: set[MessageReference] = set()
+        found: list[Message] = []
+        with closing(
+            index.recent_offsets(
+                self.since,
+                predicate=self.scope.index_predicate,
+                before_offset=source.boundary,
+            )
+        ) as offsets:
+            for row in self.indexed_records(source, offsets):
+                message = row.message
+                if message.sender in self.scope.first_names:
+                    answered.update(message.answers)
+                elif message.timestamp <= self.before and message.reference not in answered:
+                    found.append(message)
+        return tuple(found)
+
+
+@dataclass(frozen=True, slots=True)
+class SenderMessages(UnansweredUnitAsks):
+    """Everything the other side sent the commander in the window, answered or not."""
+
+    def indexed(self, index: BusPageIndex, source: WireSnapshot) -> tuple[Message, ...]:
+        with closing(
+            index.recent_offsets(
+                self.since,
+                predicate=self.scope.index_predicate,
+                before_offset=source.boundary,
+            )
+        ) as offsets:
+            return tuple(
+                row.message
+                for row in self.indexed_records(source, offsets)
+                if row.message.sender not in self.scope.first_names
+                and row.message.timestamp <= self.before
+            )
+
+
+@dataclass(frozen=True, slots=True)
+class UnansweredBurst(UnansweredUnitAsks):
+    """A sender whose messages arrive as one burst: an answer to any of its messages covers
+    everything it sent before that answer, so the commander answers a burst once, not once per
+    line. Only what it sent after the newest such answer is still open."""
+
+    def indexed(self, index: BusPageIndex, source: WireSnapshot) -> tuple[Message, ...]:
+        replied: dict[MessageReference, float] = {}
+        sent: list[Message] = []
+        with closing(
+            index.recent_offsets(
+                self.since,
+                predicate=self.scope.index_predicate,
+                before_offset=source.boundary,
+            )
+        ) as offsets:
+            for row in self.indexed_records(source, offsets):
+                message = row.message
+                if message.sender in self.scope.first_names:
+                    for answered in message.answers:
+                        replied[answered] = max(replied.get(answered, 0.0), message.timestamp)
+                else:
+                    sent.append(message)
+        covered = max((replied[m.reference] for m in sent if m.reference in replied), default=0.0)
+        return tuple(m for m in sent if covered < m.timestamp <= self.before)