diff --git a/plans/commander-native-plan.md b/plans/commander-native-plan.md
index 3fff12ac45..f9fcec6bd9 100644
--- a/plans/commander-native-plan.md
+++ b/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).
diff --git a/src/agent_comms/active_route.py b/src/agent_comms/active_route.py
index fd5dd0cb85..f861031e77 100644
--- a/src/agent_comms/active_route.py
+++ b/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,
diff --git a/src/agent_comms/commander/config.py b/src/agent_comms/commander/config.py
index dbca5bdbd0..1f237dd4ef 100644
--- a/src/agent_comms/commander/config.py
+++ b/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()
 
diff --git a/src/agent_comms/commander/order.py b/src/agent_comms/commander/order.py
index 2ad5da2e8d..d07f583004 100644
--- a/src/agent_comms/commander/order.py
+++ b/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)
 
 
diff --git a/src/agent_comms/commander/questions.py b/src/agent_comms/commander/questions.py
new file mode 100644
index 0000000000..1d172da33e
--- /dev/null
+++ b/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)
diff --git a/src/agent_comms/commander/standing.py b/src/agent_comms/commander/standing.py
index 2dbc06a464..de986e3c93 100644
--- a/src/agent_comms/commander/standing.py
+++ b/src/agent_comms/commander/standing.py
@@ -3,11 +3,9 @@
 from __future__ import annotations
 
 import time
-from abc import abstractmethod
 from collections.abc import Iterable, Sequence
 from contextlib import closing
 from dataclasses import dataclass, field
-from pathlib import Path
 from typing import ClassVar, Self
 
 from agent_comms.build_identity import RecordedBuildIdentity
@@ -17,7 +15,6 @@
 from agent_comms.commander_heartbeat import CommanderHeartbeat
 from agent_comms.comms import Comms
 from agent_comms.goal_actions import ReissuedOrderGoalAction
-from agent_comms.goal_presentation import BlockedExecution
 from agent_comms.goals import Goal
 from agent_comms.message_bus import WatchPass
 from agent_comms.message_page import MessageReadRequest, PageBounds
@@ -29,6 +26,7 @@
 from agent_comms.recorded_time import RecordedEventTime
 from agent_comms.standing_rules import (
     DatedStandingRule,
+    IndependentStandingRule,
     OwedAnswer,
     ReasonedStandingRule,
     StandingRule,
@@ -36,15 +34,15 @@
 )
 from agent_comms.storage_rules import OrderCheckoutWork
 from agent_comms.threads import Thread
-from agent_comms.todos import Todo
+from agent_comms.todos import PullRequests, Todo
 from agent_comms.turn_digests import OrderEditCountRead, TurnDigestRead, TurnsSince
+from agent_comms.ui_model.unit_state import UnitState
 from agent_comms.wire_log import WireSnapshot
-from agent_comms.work_branches import WorkBranches
 from agent_comms.work_dispatch import ReusedWorkDispatch, WorkDispatch
 
 
 @dataclass(frozen=True)
-class InstallDue(UndatedStandingRule):
+class InstallDue(IndependentStandingRule, UndatedStandingRule):
     """A configured main not contained in the installed build needs an install once its
     oldest uninstalled merge has stood for the settle window.
 
@@ -82,77 +80,7 @@ def line(self) -> str:
 
 
 @dataclass(frozen=True)
-class HolderCannotRun(UndatedStandingRule, declared_name=None):
-    unit: str
-    orders: tuple[str, ...]
-
-    @classmethod
-    def scan_owner(cls) -> type[StandingRule]:
-        return HolderCannotRun
-
-    @staticmethod
-    @abstractmethod
-    def stalled(comms: Comms, thread: Thread) -> bool: ...
-
-    @classmethod
-    @abstractmethod
-    def found(cls, comms: Comms, thread: Thread, orders: tuple[str, ...]) -> Self: ...
-
-    @classmethod
-    def violations(cls, comms: Comms, commander: str) -> tuple[Self, ...]:
-        registry = comms.registry.snapshot()
-        found: list[Self] = []
-        members = cls.members_with(cls)
-        for owner, todos in cls.holders(comms, commander).items():
-            thread = registry.require(owner.name)
-            if thread.incarnation != owner:
-                raise ValueError("Held order owner incarnation changed")
-            orders = tuple(todo.id for todo in todos)
-            found.extend(member.found(comms, thread, orders)
-                         for member in members if member.stalled(comms, thread))
-        return tuple(found)
-
-
-@dataclass(frozen=True)
-class Detached(HolderCannotRun):
-    unread: int = field(compare=False)
-
-    @staticmethod
-    def stalled(comms: Comms, thread: Thread) -> bool:
-        return not comms.registry.status(thread.name).receiving
-
-    @classmethod
-    def found(cls, comms: Comms, thread: Thread, orders: tuple[str, ...]) -> Self:
-        return cls(thread.name, orders, comms.bus.pending_count(thread.name))
-
-    def line(self) -> str:
-        return (f"DETACHED {self.unit}: {self.unread} unread, orders {', '.join(self.orders)}"
-                " -> wake it or reassign")
-
-
-@dataclass(frozen=True)
-class GoalCannotRun(UndatedStandingRule):
-    """Report blocked work without rewriting the unit's goal or orders."""
-
-    unit: str
-    orders: tuple[str, ...]
-    execution: BlockedExecution
-
-    @classmethod
-    def violations(cls, comms: Comms, commander: str) -> tuple[Self, ...]:
-        return tuple(
-            cls(owner.name, tuple(todo.id for todo in todos), block)
-            for owner, todos in cls.holders(comms, commander).items()
-            for view in (comms.goals.goal_snapshot(owner.name),)
-            for block in view.execution.unit_blocks()
-        )
-
-    def line(self) -> str:
-        return self.execution.unit_block_line(self.unit, self.orders)
-
-
-@dataclass(frozen=True)
-class RequestStalled(UndatedStandingRule):
+class RequestStalled(IndependentStandingRule, UndatedStandingRule):
     goal: Goal
 
     @classmethod
@@ -192,65 +120,7 @@ def indexed(self, index: BusPageIndex, source: WireSnapshot) -> tuple[Message, .
         return tuple(reply for reply in replies if asked.intersection(reply.answers))
 
 
-@dataclass(frozen=True, slots=True)
-class UnansweredUnitAsks(MessageReadRequest[tuple[Message, ...]]):
-    scope: RepliesDisplayScope
-    since: float
-    before: float
-    empty: ClassVar[tuple[Message, ...]] = ()
-
-    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)
-
-
-class PlanRule(UndatedStandingRule, declared_name=None):
+class PlanRule(IndependentStandingRule, UndatedStandingRule, declared_name=None):
     """A rule read from one look at the plans and the commander's orders."""
 
     @staticmethod
@@ -320,32 +190,25 @@ def line(self) -> str:
 
 @dataclass(frozen=True)
 class UnitAskUnanswered(OwedAnswer, UndatedStandingRule):
+    """Per-message gate entries derived from the unit frame, not another unit scan."""
+
     unit: str
-    seconds: ClassVar[int] = 5 * 60
 
-    @staticmethod
-    def asks(comms: Comms, commander: str, before: float) -> tuple[Message, ...]:
-        """Holders' messages to the commander, sent by ``before``, that no commander reply answers."""
-        held = StandingRule.holders(comms, commander)
-        if not held:
-            return ()
-        snapshot = comms.registry.snapshot()
-        peers = frozenset(name for owner in held for name in snapshot.names_for(owner.name))
-        since = min(todo.order.checked().at.recorded_at("A held order")
-                    for todos in held.values() for todo in todos)
-        return comms.bus.log.read_page(UnansweredUnitAsks(
-            RepliesDisplayScope(snapshot.names_for(commander), peers), since, before,
-        ))
+    @classmethod
+    def scan_observed(cls, comms: Comms, commander: str,
+                      observed: Sequence[UnitState]) -> tuple[Self, ...]:
+        return tuple(cls(message, state.unit.name) for state in observed for message in state.owed)
 
     @classmethod
     def violations(cls, comms: Comms, commander: str) -> tuple[Self, ...]:
-        snapshot = comms.registry.snapshot()
-        return tuple(cls(message, snapshot.canonical_name(message.sender))
-                     for message in cls.asks(comms, commander, time.time() - cls.seconds))
+        return tuple(cls(message, source.thread.name)
+                     for source in UnitState.observations(comms, commander, PullRequests())
+                     for message in source.due_questions)
 
     @classmethod
     def open_messages(cls, comms: Comms, commander: str) -> tuple[Message, ...]:
-        return cls.asks(comms, commander, time.time())
+        return tuple(message for source in UnitState.observations(comms, commander, PullRequests())
+                     for message in source.questions)
 
     def line(self) -> str:
         age = max(0, int((time.time() - self.message.timestamp) // 60))
@@ -354,7 +217,7 @@ def line(self) -> str:
 
 
 @dataclass(frozen=True)
-class UnownedWheelSpin(UndatedStandingRule):
+class UnownedWheelSpin(IndependentStandingRule, UndatedStandingRule):
     """A unit holds no next order while the board has unowned rows; Core hands the row to it
     without a commander turn, restarting the row's review window at the handoff."""
 
@@ -484,30 +347,7 @@ def heartbeat_line(self, current: Sequence[WaitingUnownedWheelSpin]) -> str:
 
 
 @dataclass(frozen=True)
-class CheckoutUnreadable(UndatedStandingRule):
-    """A holder's checkout git cannot open: that unit's state, while every other unit reads on."""
-
-    unit: str
-    checkout: Path
-    error: str = field(compare=False)
-
-    @classmethod
-    def violations(cls, comms: Comms, commander: str) -> tuple[Self, ...]:
-        storage = comms.registry.storage_rule
-        holders = sorted({todo.held().owner for todo in cls.pending_todos(comms, commander)})
-        return tuple(
-            cls(holder, path, error)
-            for holder in holders
-            for path in storage.checkout_paths(comms.registry.require(holder))
-            for error in WorkBranches(path).read_checkout().errors()
-        )
-
-    def line(self) -> str:
-        return (f"CHECKOUT UNREADABLE: {self.unit} {self.checkout}: {self.error}"
-                " -> order the unit to repair its checkout; its work stays retained")
-
-@dataclass(frozen=True)
-class HeartbeatSilent(DatedStandingRule):
+class HeartbeatSilent(IndependentStandingRule, DatedStandingRule):
     """A commander the heartbeat serves whose newest beat is older than the cadence allows.
 
     The heartbeat beats on every pass, so its silence is the failure; this rule reads the
diff --git a/src/agent_comms/commander/vibe_check.py b/src/agent_comms/commander/vibe_check.py
index 47d76cd1e4..29ccbd7b6d 100644
--- a/src/agent_comms/commander/vibe_check.py
+++ b/src/agent_comms/commander/vibe_check.py
@@ -16,7 +16,8 @@
 
 from agent_comms.commander.capacity import Load
 from agent_comms.commander.config import CommanderConfig
-from agent_comms.commander.standing import AnswersTo, PlanStalled, SenderMessages, UnansweredBurst
+from agent_comms.commander.questions import SenderMessages, UnansweredBurst
+from agent_comms.commander.standing import AnswersTo, PlanStalled
 from agent_comms.comms import Comms
 from agent_comms.declared_family import DeclaredFamily
 from agent_comms.goal_actions import (
@@ -27,7 +28,12 @@
 from agent_comms.messages import Message
 from agent_comms.read_basis import RepliesDisplayScope
 from agent_comms.service_sender import ServiceSender
-from agent_comms.standing_rules import OwedAnswer, StandingRule, UndatedStandingRule
+from agent_comms.standing_rules import (
+    IndependentStandingRule,
+    OwedAnswer,
+    StandingRule,
+    UndatedStandingRule,
+)
 
 
 class EscalationSender(ServiceSender):
@@ -140,7 +146,7 @@ class AnswersOwedDegradation(Degradation, BlocksRelease):
     def found(cls, comms: Comms, commander: str) -> tuple[Sign, ...]:
         return tuple(
             Sign(cls, (rule.cited,), rule.line())
-            for rule in OwedAnswer.broken(comms, commander)
+            for rule in OwedAnswer.overdue(comms, commander)
             if not isinstance(rule, Escalation)
         )
 
@@ -180,7 +186,7 @@ def found(cls, comms: Comms, commander: str) -> tuple[Sign, ...]:
 
 
 @dataclass(frozen=True)
-class VibeCheck(UndatedStandingRule):
+class VibeCheck(IndependentStandingRule, UndatedStandingRule):
     """Each degraded sign breaks it once: the heartbeat enforces a rule when it newly breaks, so a
     sign that persists stops the commander once, and a new message, a newly stalled phase or a
     doubled stall stops it again. Pings are decisions, each once."""
@@ -267,7 +273,7 @@ def quoted(cls, body: str) -> bool:
 
 
 @dataclass(frozen=True)
-class Chokehold(UndatedStandingRule):
+class Chokehold(IndependentStandingRule, UndatedStandingRule):
     """A vibe check ignored past its window: every unit under the commander is paused and the
     commander is told, in a burst it cannot skim, what brings them back."""
 
@@ -332,7 +338,7 @@ def line(self) -> str:
 
 
 @dataclass(frozen=True)
-class ChokeholdAnswer(UndatedStandingRule, declared_name=None):
+class ChokeholdAnswer(IndependentStandingRule, UndatedStandingRule, declared_name=None):
     """The commander's newest answer to a chokehold, once every chokehold message is answered."""
 
     answer: Message
diff --git a/src/agent_comms/commander_heartbeat.py b/src/agent_comms/commander_heartbeat.py
index 4cfb9a28c6..728d8afa9a 100644
--- a/src/agent_comms/commander_heartbeat.py
+++ b/src/agent_comms/commander_heartbeat.py
@@ -11,6 +11,8 @@
 from agent_comms.standing_rules import StandingRule
 from agent_comms.thread_identity import ThreadIncarnation
 from agent_comms.threads import Thread
+from agent_comms.todos import PullRequests
+from agent_comms.ui_model.unit_state import UnitState
 
 if TYPE_CHECKING:
     from agent_comms.comms import Comms
@@ -20,45 +22,40 @@
 
 @dataclass
 class CommanderReview:
-    """One commander's handled observations, its last changed beat and next review position.
+    """One commander's handled observations and its last changed beat.
 
     Seq 0 precedes every published message: no changed beat has been sent yet.
     """
 
-    violations: frozenset[StandingRule]
+    rules: frozenset[StandingRule]
+    states: frozenset[UnitState] = frozenset()
     changed: int = 0
-    turn: int = 0
 
     def wake(self, comms: Comms, sender: str, commander: Thread,
-             rules: tuple[StandingRule, ...]) -> None:
+             rules: tuple[StandingRule, ...], states: tuple[UnitState, ...]) -> None:
         """Every pass beats, so silence is a failure; an unchanged beat is one short line."""
-        found = frozenset(rules)
-        previous = self.violations
-        if found == previous and self.changed:
-            comms.messaging.send(sender, commander.name, (
-                f"COMMANDER HEARTBEAT: unchanged since seq {self.changed},"
-                f" {len(found)} violations."))
+        found_rules, found_states = frozenset(rules), frozenset(states)
+        actionable = (*rules, *(state for state in states if state.attention()))
+        old_rules = self.rules
+        previous = self.rules | self.states
+        if found_rules == self.rules and found_states == self.states and self.changed:
+            comms.messaging.send(sender, commander.name,
+                                 f"COMMANDER HEARTBEAT: unchanged since seq {self.changed},"
+                                 f" {len(actionable)} violations.")
             return
         # Mark the observation before any effect; an uncertain publication is not replayed.
-        self.violations = found
+        self.rules, self.states = found_rules, found_states
         rejected = StandingRule.enforce_all(
-            comms, commander.name, (rule for rule in rules if rule not in previous)
+            comms, commander.name, (rule for rule in actionable if rule not in previous)
         )
         # A checked rejection has no effects and must not suppress a fresh observation.
-        self.violations = found - rejected
-        lines = ["COMMANDER HEARTBEAT: standing-rule violations changed."]
-        owed = comms.goals.todos.held_for_commander(commander.incarnation)
-        units = sorted({todo.held().owner for todo in owed})
-        if units:
-            unit = units[self.turn % len(units)]
-            orders = ", ".join(todo.id for todo in owed if todo.held().owner == unit)
-            lines.append(f"Review next: @{unit} ({orders}); read its git state and newest turn.")
-            self.turn += 1
-        lines.extend(StandingRule.heartbeat_changes(
-            previous, tuple(rule for rule in rules if rule not in rejected)
-        ))
-        lines.extend(rule.rejected_line() for rule in rules if rule in rejected)
-        if not found:
+        self.rules -= rejected
+        self.states -= rejected
+        lines = ["COMMANDER HEARTBEAT: unit states and standing rules changed."]
+        lines.extend(state.line() for state in states)
+        lines.extend(StandingRule.heartbeat_changes(old_rules, self.rules))
+        lines.extend(rule.rejected_line() for rule in actionable if rule in rejected)
+        if not actionable:
             lines.append("Standing rules clear.")
         self.changed = comms.messaging.send(sender, commander.name, "\n".join(lines)).message.seq
 
@@ -68,6 +65,15 @@ class CommanderObservation:
     thread: Thread
     merged: tuple[Todo, ...]
     rules: tuple[StandingRule, ...]
+    states: tuple[UnitState, ...]
+
+    @classmethod
+    def read(cls, comms: Comms, thread: Thread) -> CommanderObservation:
+        pulls = PullRequests(StandingRule.todos(comms, thread.name))
+        states = UnitState.current(comms, thread.name, pulls)
+        rules = StandingRule.broken(comms, thread.name, observed=states)
+        merged = tuple(todo for state in states for todo in state.merged_orders(comms, pulls))
+        return cls(thread, merged, rules, states)
 
 
 @dataclass
@@ -89,11 +95,7 @@ def commanders(cls, registry: Registration, goals: Goals) -> tuple[Thread, ...]:
 
     def observe(self, comms: Comms) -> tuple[CommanderObservation, ...]:
         return tuple(
-            CommanderObservation(
-                thread,
-                comms.goals.todos.merged_orders(thread.incarnation, comms.registry),
-                StandingRule.broken(comms, thread.name),
-            )
+            CommanderObservation.read(comms, thread)
             for thread in self.commanders(comms.registry, comms.goals)
         )
 
@@ -107,4 +109,4 @@ def pass_body(self, comms: Comms, observed: tuple[CommanderObservation, ...]) ->
         for item in observed:
             comms.goals.todos.record_merges(item.merged)
             review = self.reviews.setdefault(item.thread.incarnation, CommanderReview(frozenset()))
-            review.wake(comms, self.sender, item.thread, item.rules)
+            review.wake(comms, self.sender, item.thread, item.rules, item.states)
diff --git a/src/agent_comms/commander_parity/deletion.py b/src/agent_comms/commander_parity/deletion.py
index ce2a1f6073..194bfeaa6c 100644
--- a/src/agent_comms/commander_parity/deletion.py
+++ b/src/agent_comms/commander_parity/deletion.py
@@ -15,6 +15,7 @@
 
 from ..change_report import FileLine
 from ..child_process import CountPrinted, ExitedOutcome
+from ..commander.config import CommanderConfig
 from ..commander.order import OrderFile
 from ..field_codec import FieldCodec
 from ..goal_actions import CreateProblemGoalAction
@@ -22,6 +23,7 @@
 from ..progress_claims import MergedClaim
 from ..recorded_time import RecordedEventTime
 from ..repo import Repo
+from ..storage_rules import StorageRule
 from ..todos import (
     CheckedCommitDelivery,
     CommitDelivery,
@@ -85,10 +87,10 @@ def read(cls) -> LifecycleObservation:
                 prefix="parity-deletion-", dir=rule.checkout_base
             ) as checkouts:
                 root = original.commander.root
-                (root / "storage.json").write_text(
+                StorageRule.path_in(root).write_text(
                     json.dumps(FieldCodec.encode(replace(rule, checkout_base=Path(checkouts))))
                 )
-                (root / "commander.json").write_text(original.commander.model_dump_json())
+                CommanderConfig.path_in(root).write_text(original.commander.model_dump_json())
                 run = replace(original, commander=MeasuredCommander.read_root(root))
                 return cls.exercise(run)
 
diff --git a/src/agent_comms/commander_parity/measurements.py b/src/agent_comms/commander_parity/measurements.py
index 7ebbf4f03c..c4f8dc2efb 100644
--- a/src/agent_comms/commander_parity/measurements.py
+++ b/src/agent_comms/commander_parity/measurements.py
@@ -12,6 +12,8 @@
 from agent_comms.comms import Comms
 from agent_comms.standing_rules import DiskLow, StandingRule
 from agent_comms.storage_rules import StorageViolation
+from agent_comms.todos import PullRequests
+from agent_comms.ui_model.unit_state import UnitState
 
 
 @dataclass(frozen=True)
@@ -36,12 +38,15 @@ def describe(self) -> str:
 
     @classmethod
     def read(cls, owner: type[StandingRule]) -> ParityComparison:
-        method = owner.violations
+        method = owner.scan_observed
         if not isinstance(method, MethodType):
-            return UnverifiedParity(f"{owner.__qualname__}.violations is not a bound classmethod")
+            return UnverifiedParity(
+                f"{owner.__qualname__}.scan_observed is not a bound classmethod"
+            )
         code = method.__func__.__code__
         config = CommanderConfig.read(CommanderConfig.path())
         comms = Comms(config.root)
+        states = UnitState.current(comms, config.commander, PullRequests())
         before = state_files(config.state)
         observed: list[tuple[StandingRule, ...]] = []
 
@@ -55,12 +60,12 @@ def observe(_code: CodeType, _offset: int, value: tuple[StandingRule, ...]) -> N
                 monitoring.PROFILER_ID, monitoring.events.PY_RETURN, observe
             )
             monitoring.set_local_events(monitoring.PROFILER_ID, code, monitoring.events.PY_RETURN)
-            current = owner.broken(comms, config.commander)
+            current = owner.broken(comms, config.commander, observed=states)
         finally:
             monitoring.free_tool_id(monitoring.PROFILER_ID)
         if len(observed) != 1:
             return UnverifiedParity(
-                f"Expected one {owner.__qualname__}.violations call: {len(observed)}"
+                f"Expected one {owner.__qualname__}.scan_observed call: {len(observed)}"
             )
         native = tuple(rule for rule in observed[0] if rule.broke_after(0.0))
         return cls(
@@ -86,7 +91,7 @@ def describe(self) -> str:
 
     @classmethod
     def read(cls) -> StorageComparison:
-        return cls(RuleComparison.read(StorageViolation), RuleComparison.read(DiskLow))
+        return cls(RuleComparison.read(StorageViolation.scan_owner()), RuleComparison.read(DiskLow))
 
 
 def measured_load() -> ParityComparison:
diff --git a/src/agent_comms/commander_parity/rows.py b/src/agent_comms/commander_parity/rows.py
index 511bb1a3d1..1290b52ea1 100644
--- a/src/agent_comms/commander_parity/rows.py
+++ b/src/agent_comms/commander_parity/rows.py
@@ -197,7 +197,7 @@ class StorageHygieneParityRow(ParityRow):
     number = 19
     capability = "Storage hygiene"
     current = "`StandingRule.broken` storage delivery (formerly `storage.py`)"
-    native = "`StorageViolation.violations`, `DiskLow.violations`; live `StorageRule` thresholds"
+    native = "`StorageViolation.scan_observed`, `DiskLow.violations`; live `StorageRule` thresholds"
     phases = (ControlPlanePhase,)
 
     @classmethod
diff --git a/src/agent_comms/goal_presentation.py b/src/agent_comms/goal_presentation.py
index 20584dac9e..fa76b14ad5 100644
--- a/src/agent_comms/goal_presentation.py
+++ b/src/agent_comms/goal_presentation.py
@@ -80,6 +80,9 @@ def details_heading(self, heading: str) -> str:
     def waiting_presentation(self, ordinary: ThreadPresentation) -> ThreadPresentation:
         return ordinary
 
+    def runnable(self) -> bool:
+        return False
+
     def standby(self) -> bool:
         return False
 
@@ -162,6 +165,9 @@ def toggle(self) -> type[GoalAction]:
 class RunnableExecution(ActiveExecution):
     label = "Active"
 
+    def runnable(self) -> bool:
+        return True
+
 
 @dataclass(frozen=True)
 class RunningExecution(ActiveExecution):
diff --git a/src/agent_comms/messages.py b/src/agent_comms/messages.py
index c7bf2bb939..503d3c5ff8 100644
--- a/src/agent_comms/messages.py
+++ b/src/agent_comms/messages.py
@@ -108,7 +108,11 @@ def decode_message(cls, data: Mapping) -> Message:
         return cast(Message, compile_decoder(Message | ClaimMessage).decode(data))
 
 
-class QuestionKind(SendableMessageKind):
+class NeedsAnswerKind(SendableMessageKind, declared_name=None):
+    """Questions, alerts and handoffs ask their recipient to act."""
+
+
+class QuestionKind(NeedsAnswerKind):
     pass
 
 
@@ -116,7 +120,7 @@ class AckKind(SendableMessageKind):
     pass
 
 
-class HandoffKind(SendableMessageKind):
+class HandoffKind(NeedsAnswerKind):
     """A handoff declares its evidence; completion claims carry references."""
 
     CLAIMS = ("fixed", "single owner", "no other callers")
@@ -137,7 +141,7 @@ def require_publishable(cls, message: Message, worktree: Path) -> None:
         )
 
 
-class AlertKind(SendableMessageKind):
+class AlertKind(NeedsAnswerKind):
     pass
 
 
diff --git a/src/agent_comms/skills/comms-commander/SKILL.md b/src/agent_comms/skills/comms-commander/SKILL.md
index 89f82b997b..9674e6f918 100644
--- a/src/agent_comms/skills/comms-commander/SKILL.md
+++ b/src/agent_comms/skills/comms-commander/SKILL.md
@@ -72,13 +72,7 @@ Priority: the owner's latest message; a broken install, lost or replayed input,
 
 ## One rotation tick
 
-Pick the next unit. Read its actual state, not its message: worktree, `git log -1`, `git status`, the diff, its last turns in tool-watch. Decide:
-- **Moving:** a fresh commit that deletes or fixes the right thing. Leave it.
-- **Spinning:** repeated send-backs, no commit for 20 minutes on exact edits, tests rerun as investigation. Write the exact target shape into the order.
-- **Stuck:** no tool call for 3 minutes while holding an order. Order it.
-- **Stalled on nothing:** ended its turn holding an order with a question it could answer itself, a summary, a plan, "ready to proceed", or a wait on something already given. Send the next concrete step the same tick, with a strike (rules/send-backs-and-strikes.md).
-- **Idle:** give it the top unowned plan step if your load has room to review it; otherwise leave it stopped.
-- **Adding slop:** send it back with the result it violates.
+Read the heartbeat's unit cards in their urgency order. Each `UnitState` member owns its classification and default action. Read the selected unit's diff and newest turn to judge whether its work is correct; send back added slop with the result it violates.
 
 One line per unit touched, then the next unit.
 
diff --git a/src/agent_comms/standing_rules.py b/src/agent_comms/standing_rules.py
index f2867f8d62..f88d1d7aa3 100644
--- a/src/agent_comms/standing_rules.py
+++ b/src/agent_comms/standing_rules.py
@@ -28,12 +28,14 @@
     from agent_comms.storage_rules import MemoryUse, StorageRule
     from agent_comms.thread_identity import ThreadIncarnation
     from agent_comms.todos import Todo
+    from agent_comms.ui_model.unit_state import UnitState
 
 
 class StandingRule(DeclaredFamily, affix="StandingRule", discovery_package="agent_comms.commander"):
     @classmethod
     @abstractmethod
-    def violations(cls, comms: Comms, commander: str) -> tuple[Self, ...]: ...
+    def scan_observed(cls, comms: Comms, commander: str,
+                      observed: Sequence[UnitState]) -> tuple[Self, ...]: ...
 
     @abstractmethod
     def line(self) -> str: ...
@@ -41,6 +43,9 @@ def line(self) -> str: ...
     @abstractmethod
     def broke_after(self, after: float) -> bool: ...
 
+    def attention(self) -> bool:
+        return True
+
     def rejected_line(self) -> str:
         return f"Not applied (goal changed): {self.line()}"
 
@@ -98,9 +103,9 @@ def pending_todos(comms: Comms, commander: str) -> tuple[Todo, ...]:
 
     @staticmethod
     def holders(comms: Comms, commander: str) -> dict[ThreadIncarnation, list[Todo]]:
-        """Every unit holding a pending order, with its orders."""
+        """Every unit holding an open order, with its orders."""
         held: dict[ThreadIncarnation, list[Todo]] = {}
-        for todo in StandingRule.pending_todos(comms, commander):
+        for todo in StandingRule.todos(comms, commander):
             held.setdefault(todo.held().owner_identity, []).append(todo)
         return held
 
@@ -110,11 +115,26 @@ def scan_owner(cls) -> type[StandingRule]:
         return cls
 
     @classmethod
-    def broken(cls, comms: Comms, commander: str, after: float = 0.0) -> tuple[Self, ...]:
-        """Read each inherited scan owner once, preserving the requested member scope."""
+    def broken(cls, comms: Comms, commander: str, after: float = 0.0, *,
+               observed: Sequence[UnitState]) -> tuple[Self, ...]:
+        """Read each owner once against the supplied frame, preserving the requested family."""
         owners = dict.fromkeys(rule.scan_owner() for rule in cls.members_with(cls))
-        found = (v for owner in owners for v in owner.violations(comms, commander))
-        return tuple(rule for rule in found if isinstance(rule, cls) and rule.broke_after(after))
+        return tuple(rule for owner in owners
+                     for rule in owner.scan_observed(comms, commander, observed)
+                     if isinstance(rule, cls) and rule.attention() and rule.broke_after(after))
+
+
+class IndependentStandingRule(StandingRule, declared_name=None):
+    """A rule whose source read does not consume the unit observation frame."""
+
+    @classmethod
+    @abstractmethod
+    def violations(cls, comms: Comms, commander: str) -> tuple[Self, ...]: ...
+
+    @classmethod
+    def scan_observed(cls, comms: Comms, commander: str,
+                      observed: Sequence[UnitState]) -> tuple[Self, ...]:
+        return cls.violations(comms, commander)
 
 
 class UndatedStandingRule(StandingRule, declared_name=None):
@@ -135,7 +155,7 @@ def broke_after(self, after: float) -> bool:
 
 
 @dataclass(frozen=True)
-class OwedAnswer(StandingRule, declared_name=None):
+class OwedAnswer(IndependentStandingRule, declared_name=None):
     """A message to the commander owed an answer past its window. Every member gates the
     commander's answering verbs until an action names it with --answers, so an answer cannot
     be held: one that must wait is sent now as "waiting on X, owner Y"."""
@@ -152,6 +172,12 @@ def cited(self) -> str:
     def open_messages(cls, comms: Comms, commander: str) -> tuple[Message, ...]:
         """This member's messages an action may answer now, inside their window or past it."""
 
+    @classmethod
+    def overdue(cls, comms: Comms, commander: str) -> tuple[Self, ...]:
+        """The message gate reads message debt, without classifying checkouts or runtimes."""
+        return tuple(rule for member in cls.members_with(cls)
+                     for rule in member.violations(comms, commander))
+
     def answered_by(self, named: tuple[Message, ...]) -> bool:
         """Whether an action naming these open messages answers this one."""
         return any(message.message_id == self.message.message_id for message in named)
@@ -175,20 +201,22 @@ class ReasonedStandingRule(StandingRule, declared_name=None):
     def reason(cls, comms: Comms, commander: str) -> str | None: ...
 
     @classmethod
-    def found(cls, why: str, comms: Comms, commander: str) -> Self:
+    def found(cls, why: str, comms: Comms, commander: str,
+              observed: Sequence[UnitState]) -> Self:
         return cls(why)
 
     @classmethod
-    def violations(cls, comms: Comms, commander: str) -> tuple[Self, ...]:
+    def scan_observed(cls, comms: Comms, commander: str,
+                      observed: Sequence[UnitState]) -> tuple[Self, ...]:
         why = cls.reason(comms, commander)
-        return (cls.found(why, comms, commander),) if why else ()
+        return (cls.found(why, comms, commander, observed),) if why else ()
 
     def line(self) -> str:
         return self.text.format(why=self.why)
 
 
 @dataclass(frozen=True)
-class HeldOrderDue(DatedStandingRule, declared_name=None):
+class HeldOrderDue(IndependentStandingRule, DatedStandingRule, declared_name=None):
     """Each held order supplies one dated condition; the rule owns its wording."""
 
     problem: str
@@ -224,52 +252,6 @@ def line(self) -> str:
                 f"order {self.order} -> read its diff and judge it")
 
 
-class FirstCallMissing(HeldOrderDue):
-    @classmethod
-    def conditions(cls, comms: Comms, commander: str) -> Iterable[tuple[Todo, EventTime]]:
-        now = time.time()
-        for todo in cls.pending_todos(comms, commander):
-            order = todo.order.checked()
-            if not order.first_call_deadline.before(now):
-                continue
-            thread = comms.registry.require(todo.held().owner)
-            last = thread.runtime.history_source(comms.root, thread.incarnation).last_tool_call()
-            if not last.since(order.at.recorded_at("An order")):
-                yield todo, order.first_call_deadline
-
-    def line(self) -> str:
-        return (f"FIRST CALL MISSING {self.problem} {self.holder}, due {self.since.clock()}, "
-                f"order {self.order} -> push the work or reassign it")
-
-
-@dataclass(frozen=True)
-class OwnerIdle(DatedStandingRule):
-    seconds: ClassVar[int] = 1200
-    holder: str
-    since: float
-
-    @classmethod
-    def violations(cls, comms: Comms, commander: str) -> tuple[Self, ...]:
-        now = time.time()
-        threads = (comms.registry.require(holder)
-                   for holder in sorted({todo.held().owner
-                                         for todo in cls.pending_todos(comms, commander)}))
-        calls = (
-            (thread.name,
-             thread.runtime.history_source(comms.root, thread.incarnation).last_tool_call())
-            for thread in threads
-        )
-        return tuple(cls(holder, last.recorded_at("A tool call") + cls.seconds)
-                     for holder, last in calls if last.before(now - cls.seconds))
-
-    def began(self) -> float:
-        return self.since
-
-    def line(self) -> str:
-        return (f"OWNER IDLE {self.holder}, since {RecordedEventTime(self.since).clock()}"
-                " -> push or reassign the work")
-
-
 @dataclass(frozen=True)
 class HuntDue(ReasonedStandingRule, DatedStandingRule):
     since: float
@@ -287,13 +269,18 @@ def reason(cls, comms: Comms, commander: str) -> str | None:
         return f"last hunt {int(age // 60)} min ago" if age > cls.seconds else None
 
     @classmethod
-    def found(cls, why: str, comms: Comms, commander: str) -> HuntDue:
+    def found(cls, why: str, comms: Comms, commander: str,
+              observed: Sequence[UnitState]) -> HuntDue:
         since = cls.hunted(comms, commander) + cls.seconds
+        readable = {unit for state in observed for unit in state.readable_units()}
+        todos = tuple(todo for todo in cls.todos(comms, commander)
+                      if todo.held().owner_identity in readable)
         return next((LocatedHuntDue(why, since, path)
-                     for path in cls.next_hunt(comms, commander)), cls(why, since))
+                     for path in cls.next_hunt(comms, commander, todos)), cls(why, since))
 
     @classmethod
-    def next_hunt(cls, comms: Comms, commander: str) -> tuple[str, ...]:
+    def next_hunt(cls, comms: Comms, commander: str,
+                  todos: Iterable[Todo]) -> tuple[str, ...]:
         """Least recently hunted hot file in the held orders' declared check repositories."""
         actor = comms.registry.require(commander).incarnation
         hunted: dict[str, float] = {}
@@ -301,7 +288,7 @@ def next_hunt(cls, comms: Comms, commander: str) -> tuple[str, ...]:
             hunted[hunt.where.file] = max(hunted.get(hunt.where.file, 0),
                                           hunt.at.recorded_at("A hunt"))
         repos = {Repo(checkout=Path(todo.order.checked().check.directory), github=todo.repo)
-                 for todo in cls.todos(comms, commander)}
+                 for todo in todos}
         candidates: list[str] = []
         for repo in repos:
             log = repo.git(repo.checkout, 'log', 'origin/main', '--since=7.days',
@@ -388,7 +375,7 @@ def reason(cls, comms: Comms, commander: str) -> str | None:
 
 
 @dataclass(frozen=True)
-class ServicePassFailing(DatedStandingRule):
+class ServicePassFailing(IndependentStandingRule, DatedStandingRule):
     """A Core service watch whose newest pass failed; the connector's log reaches nobody."""
 
     watch: str
@@ -410,7 +397,7 @@ def line(self) -> str:
                 " -> fix the owner of the failing read this wake")
 
 @dataclass(frozen=True)
-class BoardDisagrees(UndatedStandingRule):
+class BoardDisagrees(IndependentStandingRule, UndatedStandingRule):
     """An open order reads as no commit despite its linked pull request having merged."""
 
     problem: str
@@ -420,7 +407,7 @@ class BoardDisagrees(UndatedStandingRule):
     @classmethod
     def violations(cls, comms: Comms, commander: str) -> tuple[Self, ...]:
         return tuple(
-            cls(todo.id, todo.order.id, merged.sha)
+            cls(todo.id, todo.order.checked().id, merged.sha)
             for todo, merged in comms.goals.todos.board_disagreements(
                 comms.registry.require(commander).incarnation, comms.registry
             )
@@ -434,7 +421,7 @@ def line(self) -> str:
 
 
 @dataclass(frozen=True)
-class MemoryLow(UndatedStandingRule):
+class MemoryLow(IndependentStandingRule, UndatedStandingRule):
     """Under the memory threshold a cleanup owner is due, told what holds the memory."""
 
     rule: StorageRule
@@ -458,7 +445,7 @@ def line(self) -> str:
 
 
 @dataclass(frozen=True)
-class DiskLow(UndatedStandingRule):
+class DiskLow(IndependentStandingRule, UndatedStandingRule):
     """Under the cleanup threshold a cleanup owner is due; under the floor nothing allocates."""
 
     rule: StorageRule
diff --git a/src/agent_comms/storage_rules.py b/src/agent_comms/storage_rules.py
index f2263d6258..503f0ef3e9 100644
--- a/src/agent_comms/storage_rules.py
+++ b/src/agent_comms/storage_rules.py
@@ -9,7 +9,7 @@
 import subprocess
 import time
 from abc import abstractmethod
-from collections.abc import Iterator, Sequence
+from collections.abc import Iterable, Iterator, Sequence
 from dataclasses import dataclass, replace
 from enum import StrEnum
 from functools import cached_property
@@ -17,11 +17,12 @@
 from typing import TYPE_CHECKING, Annotated, ClassVar, Self
 
 from agent_comms.field_codec import FieldCodec, PathText
+from agent_comms.locked_store import DurableFile
 from agent_comms.native_tool_call import SelectedToolDenied
 from agent_comms.problem_number import ProblemNumber
 from agent_comms.repo import Repo
 from agent_comms.repository_name import RepositoryName
-from agent_comms.standing_rules import UndatedStandingRule
+from agent_comms.standing_rules import StandingRule, UndatedStandingRule
 from agent_comms.storage_room import (
     AllocationRoomWork,
     BuildRoomWork,
@@ -34,8 +35,10 @@
 if TYPE_CHECKING:
     from agent_comms.comms import Comms
     from agent_comms.registration import Registration
+    from agent_comms.thread_identity import ThreadIncarnation
     from agent_comms.threads import Thread
     from agent_comms.todos import Todo, TodoStore
+    from agent_comms.ui_model.unit_state import UnitState
 
 StoragePath = Annotated[Path, PathText]
 
@@ -70,7 +73,8 @@ class MemoryUse:
 
 
 @dataclass(frozen=True)
-class StorageRule(RoomAllocation):
+class StorageRule(RoomAllocation, DurableFile):
+    filename: ClassVar[str] = "storage.json"
     floor_gib: float = 5.0  # under this nothing that allocates may run
     cleanup_gib: float = 30.0  # under this a cleanup owner is due; it does not stop builds
     memory_gib: float = 3.0  # available memory under this makes a cleanup owner due
@@ -93,7 +97,7 @@ def __post_init__(self):
 
     @classmethod
     def load(cls, root: Path):
-        path = root / 'storage.json'
+        path = cls.path_in(root)
         return FieldCodec.decode(cls, json.loads(path.read_text()) if path.exists() else {})
 
     def initial_roots(self, name):
@@ -341,14 +345,22 @@ def scan_owner(cls) -> type[StorageViolation]:
         return __class__
 
     @classmethod
-    def violations(cls, comms: Comms, commander: str) -> tuple[Self, ...]:
-        todos = cls.todos(comms, commander)
-        units = sorted({todo.holder for todo in todos if todo.state.open and todo.holder})
+    def scan_observed(cls, comms: Comms, commander: str,
+                      observed: Sequence[UnitState]) -> tuple[Self, ...]:
+        held = StandingRule.holders(comms, commander)
+        return cls.for_units(comms, (unit for state in observed
+                                    for unit in state.readable_units() if unit in held))
+
+    @classmethod
+    def for_units(cls, comms: Comms, units: Iterable[ThreadIncarnation]) -> tuple[Self, ...]:
         registry = comms.registry
         members = cls.members_with(cls)
         found = []
-        for unit in units:
-            storage = UnitStorage.read(registry.storage_rule, registry.require(unit))
+        for unit in sorted(set(units), key=lambda unit: unit.name):
+            thread = registry.require(unit.name)
+            if thread.incarnation != unit:
+                raise ValueError("Observed unit incarnation changed")
+            storage = UnitStorage.read(registry.storage_rule, thread)
             found.extend(member(storage) for member in members if member.violated(storage))
         return tuple(found)
 
diff --git a/src/agent_comms/todos.py b/src/agent_comms/todos.py
index b72d158cf4..d2eb652245 100644
--- a/src/agent_comms/todos.py
+++ b/src/agent_comms/todos.py
@@ -16,7 +16,7 @@
 import sys
 import time
 from abc import ABC, abstractmethod
-from collections.abc import Callable, Iterator, Sequence
+from collections.abc import Callable, Iterator, Mapping, Sequence
 from contextlib import contextmanager
 from dataclasses import dataclass, field, fields, replace
 from functools import cache, cached_property, partial
@@ -71,13 +71,13 @@
     WireValue,
     boundary_decoder,
 )
-from agent_comms.goals import CheckedCommit, OrderCheck, ProblemGoalOrder
+from agent_comms.goals import CheckedCommit, GitCommandFailed, OrderCheck, ProblemGoalOrder
 from agent_comms.message_reference import MessageReference
 from agent_comms.messages import InfoKind, Message
 from agent_comms.order_runs import OrderRun
 from agent_comms.order_story import ProblemStory, StoryEvent, StoryLine, StorySource
 from agent_comms.plan_phases import OrderPhase, UnplannedPhase
-from agent_comms.private_bus_checkpoint import PublicMessages
+from agent_comms.private_bus_checkpoint import CheckedRead, PublicMessages
 from agent_comms.problem_number import ProblemNumber
 from agent_comms.progress_claims import MergedClaim, ProgressClaim, WorkingClaim
 from agent_comms.recorded_time import (
@@ -580,8 +580,13 @@ class PullRequests:
     def __init__(self, todos: Sequence[Todo] = ()) -> None:
         self._read: dict[tuple[RepositoryName, type[ListedPullRequest], str], tuple[PullRequest, ...]] = {}
         self._todos: dict[RepositoryName, dict[str, Todo]] = {}
+        self.register(todos)
+
+    def register(self, todos: Sequence[Todo]) -> Self:
+        """Register a frame's orders while preserving this read's existing GitHub facts."""
         for todo in todos:
             self._todos.setdefault(todo.repo, {})[todo.id] = todo
+        return self
 
     def listed(self, repo: RepositoryName, kind: type[ListedPullRequest],
                todo: Todo) -> tuple[PullRequest, ...]:
@@ -751,6 +756,9 @@ def readings(self, source: ProgressSource) -> DistanceReadings: ...
     @abstractmethod
     def pushes(self, pattern: str) -> Iterator[tuple[str, float]]: ...
 
+    def require_read(self) -> None:
+        """A retired checkout needs no local read; present and unread ones decide."""
+
     def checked_commits(self, order: CheckedProblemOrder, refs: tuple[str, ...],
                         *excluded: str) -> Iterator[ProgressPoint]:
         return iter(())
@@ -776,6 +784,29 @@ def pushes(self, pattern: str) -> Iterator[tuple[str, float]]:
         return iter(())
 
 
+@dataclass(frozen=True)
+class UnreadCheckout(ProgressCheckout):
+    """A failed source read is data, never a missing checkout or no progress."""
+
+    reason: str
+
+    def require_read(self) -> None:
+        raise WorkBranchError(self.reason)
+
+    def commits(self, pattern: str, refs: tuple[str, ...],
+                *excluded: str) -> Iterator[ProgressPoint]:
+        raise WorkBranchError(self.reason)
+
+    def references(self, storage: StorageRule, owner: Thread) -> tuple[str, ...]:
+        return ()
+
+    def readings(self, source: ProgressSource) -> DistanceReadings:
+        raise WorkBranchError(self.reason)
+
+    def pushes(self, pattern: str) -> Iterator[tuple[str, float]]:
+        raise WorkBranchError(self.reason)
+
+
 class PresentCheckout(ProgressCheckout):
     default_reference: ClassVar[str] = "refs/remotes/origin/HEAD"
 
@@ -786,6 +817,9 @@ def default_refs(self) -> tuple[str, ...]:
     def git(self, *args: str) -> str:
         return CheckedCommit.git(str(self.path), *args)
 
+    def require_read(self) -> None:
+        self.git("rev-parse", "--git-dir")
+
     @cached_property
     def git_refs(self) -> tuple[str, ...]:
         return tuple(self.git("for-each-ref", "--format=%(refname)",
@@ -877,8 +911,13 @@ def of(cls, todo: Todo, registry: Registration, pulls: PullRequests) -> Self:
         candidates = (Path(todo.order.checked().check.directory), *owned)
         path = next((path for path in candidates if (path / ".git").exists()),
                     next(iter(owned), candidates[0]))
-        checkout = PresentCheckout(path) if (path / ".git").exists() else MissingCheckout(path)
-        return cls(todo, checkout, pulls, checkout.references(storage, owner),
+        checkout: ProgressCheckout = (
+            PresentCheckout(path) if (path / ".git").exists() else MissingCheckout(path))
+        try:
+            references = checkout.references(storage, owner)
+        except (OSError, GitCommandFailed, WorkBranchError, subprocess.TimeoutExpired) as failure:
+            checkout, references = UnreadCheckout(path, str(failure)), ()
+        return cls(todo, checkout, pulls, references,
                    tuple(PresentCheckout(path) for path in owned if (path / ".git").exists()))
 
     @property
@@ -945,7 +984,7 @@ def from_source(cls, source: ProgressSource) -> OrderProgress:
         try:
             return next((progress for member in members
                          for progress in member.observations(source)), NoCommitProgress())
-        except (WorkBranchError, subprocess.TimeoutExpired) as unread:
+        except (OSError, GitCommandFailed, WorkBranchError, subprocess.TimeoutExpired) as unread:
             return UnreadProgress(str(unread))
 
     @property
@@ -1504,6 +1543,9 @@ def withdrawals(self, todo: Todo) -> tuple[str, ...]:
     def waiting_on_install(self, todo: Todo, read: dict[str, OrderProgress]) -> tuple[Todo, ...]:
         return ()
 
+    def commander_claims(self) -> tuple[OrderClaim, ...]:
+        return ()
+
     def closing_artifact(self) -> str:
         raise ValueError("Only a delivered order has closing evidence.")
 
@@ -1995,6 +2037,21 @@ def __post_init__(self) -> None:
         if not self.message.body.strip():
             raise TodoError("A delivery claim needs its report text.")
 
+    def original(self, source: CheckedRead) -> Message:
+        row = PublicMessages.require(source.connection, message_id=self.message.message_id)
+        return replace(self.message, seq=row.seq)
+
+    @staticmethod
+    def originals[K](
+        comms: Comms, claims: Mapping[K, tuple[OrderClaim, ...]]
+    ) -> dict[K, tuple[Message, ...]]:
+        """Every claim's original from one checked read; no claim opens no read."""
+        if not any(claims.values()):
+            return {}
+        with comms.bus.log.checked_read() as source:
+            return {key: tuple(claim.original(source) for claim in held)
+                    for key, held in claims.items()}
+
     @classmethod
     def for_order(
         cls, todo: Todo, actor: ThreadIncarnation, number: int, head: str,
@@ -2019,6 +2076,9 @@ def require_reassignment(self, todo: Todo, proposed: Assignment) -> None:
         if proposed.owner_identity != todo.holder_state.previous_assignment().owner_identity:
             self.require_transfer(todo)
 
+    def commander_claims(self) -> tuple[OrderClaim, ...]:
+        return (self.claim,)
+
     def observed_merges(self, repo: Repo, source: ProgressSource) -> tuple[MergedProblemOrder, ...]:
         return ()
 
diff --git a/src/agent_comms/tools.py b/src/agent_comms/tools.py
index aa8629b343..cd395886ea 100644
--- a/src/agent_comms/tools.py
+++ b/src/agent_comms/tools.py
@@ -40,6 +40,7 @@
 )
 from .cli_result import ToolCatalogResult
 from .commander.capacity import Load
+from .ui_model.unit_state import UnitState
 from .coordination_tables.refusals import SendRefusal
 from .declared_family import DeclaredFamily
 from .display_order import ChannelSort, ThreadSort
@@ -1028,7 +1029,9 @@ class CommsStandingRulesTool(ThreadToolRequest):
     ] = 0.0
 
     def apply(self, comms: Comms, caller: ThreadCaller) -> StandingRulesRead:
-        broken = StandingRule.broken(comms, caller.name)
+        states = UnitState.current(comms, caller.name, PullRequests())
+        broken = (*StandingRule.broken(comms, caller.name, observed=states),
+                  *(state for state in states if state.attention()))
         rejected = StandingRule.enforce_all(comms, caller.name, broken)
         return StandingRulesRead(
             tuple(rule.line() for rule in broken
diff --git a/src/agent_comms/ui_model/unit_state.py b/src/agent_comms/ui_model/unit_state.py
new file mode 100644
index 0000000000..7ddf5d030a
--- /dev/null
+++ b/src/agent_comms/ui_model/unit_state.py
@@ -0,0 +1,445 @@
+"""One derived card per unit, ordered by the family's declaration order.
+
+States are siblings, not increasingly urgent subtypes of one another. The
+StandingRule registry supplies both membership and urgency; no second roster
+or independently writable unit status is retained.
+"""
+
+from __future__ import annotations
+
+import time
+from abc import abstractmethod
+from collections.abc import Iterator, Sequence
+from dataclasses import dataclass, field
+from functools import cached_property
+from subprocess import TimeoutExpired
+from typing import TYPE_CHECKING, ClassVar, Self
+
+from agent_comms.commander.questions import UnansweredUnitAsks
+from agent_comms.field_codec import MalformedRecord
+from agent_comms.goal_presentation import GoalSnapshot
+from agent_comms.goals import GitCommandFailed
+from agent_comms.messages import Message
+from agent_comms.native_failures import NativePiUnavailable
+from agent_comms.standing_rules import StandingRule, UndatedStandingRule
+from agent_comms.thread_identity import ThreadIncarnation
+from agent_comms.thread_status import ThreadStatus
+from agent_comms.todos import (
+    MergedProgress,
+    NoCommitProgress,
+    OrderClaim,
+    OrderProgress,
+    PresentCheckout,
+    ProgressSource,
+    PullRequests,
+    SendBackVerdict,
+    Todo,
+)
+from agent_comms.transcript_events import OrderCall, TurnDigest
+from agent_comms.turn_digests import LastTurns, TurnDigestRead
+from agent_comms.work_branches import WorkBranchError
+
+if TYPE_CHECKING:
+    from agent_comms.comms import Comms
+    from agent_comms.threads import Thread
+
+
+@dataclass(frozen=True)
+class UnitObservation:
+    """Original sources shared by candidate states within this one read."""
+
+    comms: Comms
+    thread: Thread
+    todos: tuple[Todo, ...]
+    pulls: PullRequests
+    questions: tuple[Message, ...]
+    now: float
+    presence: ThreadStatus
+    working: bool
+
+    @cached_property
+    def goal(self) -> GoalSnapshot:
+        return self.comms.goals.goal_snapshot(self.thread.name)
+
+    @cached_property
+    def due_questions(self) -> tuple[Message, ...]:
+        return tuple(
+            message
+            for message in self.questions
+            if message.timestamp <= self.now - AwaitingCommander.seconds
+        )
+
+    @cached_property
+    def pending(self) -> tuple[Todo, ...]:
+        return tuple(todo for todo in self.todos if todo.pending)
+
+    @cached_property
+    def turns(self) -> tuple[TurnDigest, ...]:
+        return TurnDigestRead(self.comms.root, self.thread, LastTurns(2)).read().turns
+
+    @cached_property
+    def calls(self) -> tuple[OrderCall, ...]:
+        calls: dict[str, OrderCall] = {}
+        for turn in reversed(self.turns):
+            for event in turn.events:
+                event.record_call(calls, self.since)
+        return tuple(calls.values())
+
+    @cached_property
+    def since(self) -> float:
+        return min(todo.order.checked().at.recorded_at("A held order") for todo in self.pending)
+
+    @cached_property
+    def sources(self) -> tuple[ProgressSource, ...]:
+        return tuple(
+            ProgressSource.of(todo, self.comms.registry, self.pulls) for todo in self.todos
+        )
+
+    def require_read(self) -> None:
+        if self.todos:
+            for path in self.comms.registry.storage_rule.checkout_paths(self.thread):
+                PresentCheckout(path).require_read()
+        for source in self.sources:
+            source.checkout.require_read()
+
+    @property
+    def orders(self) -> tuple[str, ...]:
+        return tuple(todo.id for todo in self.todos)
+
+
+class NeedsCommanderAttention:
+    """Urgent siblings share their ping behavior without claiming each other's state."""
+
+    def attention(self) -> bool:
+        return True
+
+
+@dataclass(frozen=True)
+class UnitState(UndatedStandingRule, declared_name=None):
+    unit: ThreadIncarnation
+    orders: tuple[str, ...]
+    presence: ThreadStatus
+    source: UnitObservation = field(compare=False, repr=False)
+    action: ClassVar[str]
+
+    @classmethod
+    def worst_first(cls) -> tuple[type[UnitState], ...]:
+        return UnitState.members_with(UnitState)
+
+    @classmethod
+    def scan_owner(cls) -> type[StandingRule]:
+        return UnitState
+
+    @classmethod
+    def scan_observed(
+        cls, comms: Comms, commander: str, observed: Sequence[UnitState]
+    ) -> tuple[Self, ...]:
+        # States are already read; ordinary scans consume them rather than emit them again.
+        return ()
+
+    @classmethod
+    def read(cls, source: UnitObservation) -> UnitState:
+        try:
+            return next(state for member in cls.worst_first() for state in member.derive(source))
+        except (
+            OSError,
+            GitCommandFailed,
+            WorkBranchError,
+            TimeoutExpired,
+            NativePiUnavailable,
+            MalformedRecord,
+        ) as failure:
+            return Unreadable(
+                source.thread.incarnation, source.orders, source.presence, source, str(failure)
+            )
+
+    @classmethod
+    def observations(
+        cls, comms: Comms, commander: str, pulls: PullRequests
+    ) -> tuple[UnitObservation, ...]:
+        snapshot, now = comms.registry.snapshot(), time.time()
+        held = StandingRule.holders(comms, commander)
+        for owner in held:
+            if snapshot.require(owner.name).incarnation != owner:
+                raise ValueError("Held order owner incarnation changed")
+        requests = pulls.register(tuple(todo for todos in held.values() for todo in todos))
+        asks = UnansweredUnitAsks.for_orders(
+            comms, commander, StandingRule.todos(comms, commander), now
+        )
+        # Claim originals come from one read here, so no unit's questions read can fail alone.
+        claims = OrderClaim.originals(
+            comms,
+            {
+                owner: tuple(claim for todo in todos for claim in todo.order.commander_claims())
+                for owner, todos in held.items()
+            },
+        )
+        readings = tuple(
+            UnitObservation(
+                comms,
+                thread,
+                tuple(held.get(thread.incarnation, ())),
+                requests,
+                (
+                    *(
+                        message
+                        for message in asks
+                        if snapshot.canonical_name(message.sender) == thread.name
+                    ),
+                    *claims.get(thread.incarnation, ()),
+                ),
+                now,
+                snapshot.status(thread.name),
+                comms.agents.is_working(thread.incarnation, snapshot),
+            )
+            for thread in snapshot.threads.values()
+        )
+        candidates = tuple(reading for reading in readings if reading.thread.name != commander)
+        eligible = set(held)
+        eligible.update(
+            reading.thread.incarnation
+            for reading in candidates
+            if reading.thread.parent.named(commander)
+        )
+        eligible.update(
+            reading.thread.incarnation
+            for reading in candidates
+            if reading.thread.role.executor(reading.thread) and reading.presence.receiving
+        )
+        return tuple(reading for reading in readings if reading.thread.incarnation in eligible)
+
+    @classmethod
+    def current(cls, comms: Comms, commander: str, pulls: PullRequests) -> tuple[UnitState, ...]:
+        states = tuple(cls.read(source) for source in cls.observations(comms, commander, pulls))
+        return tuple(
+            state for member in cls.worst_first() for state in states if isinstance(state, member)
+        )
+
+    @classmethod
+    @abstractmethod
+    def derive(cls, source: UnitObservation) -> Iterator[Self]: ...
+
+    @property
+    def owed(self) -> tuple[Message, ...]:
+        return self.source.due_questions
+
+    def readable_units(self) -> tuple[ThreadIncarnation, ...]:
+        return (self.unit,)
+
+    def attention(self) -> bool:
+        return False
+
+    def merged_orders(self, comms: Comms, pulls: PullRequests) -> tuple[Todo, ...]:
+        return tuple(
+            closed
+            for problem in self.orders
+            for todo in (comms.goals.todos.get(problem),)
+            for closed in todo.order.closed_by_merge(todo, comms.registry, pulls)
+        )
+
+    @property
+    def default_action(self) -> str:
+        return self.action
+
+    def detail(self) -> str:
+        return f"orders {', '.join(self.orders)}"
+
+    def line(self) -> str:
+        return f"{type(self).__name__} {self.unit.name}: {self.detail()} -> {self.default_action}"
+
+
+class ConditionalState(UnitState, declared_name=None):
+    """Abstract card construction shared by siblings, not a separate unit state."""
+
+    @classmethod
+    def derive(cls, source: UnitObservation) -> Iterator[Self]:
+        if cls.matches(source):
+            yield cls(source.thread.incarnation, source.orders, source.presence, source)
+
+    @classmethod
+    @abstractmethod
+    def matches(cls, source: UnitObservation) -> bool: ...
+
+
+class PendingWorkState(ConditionalState, declared_name=None):
+    """These sibling states require pending work; declare that condition once."""
+
+    @classmethod
+    def derive(cls, source: UnitObservation) -> Iterator[Self]:
+        if source.pending:
+            yield from super().derive(source)
+
+
+@dataclass(frozen=True)
+class Unreadable(NeedsCommanderAttention, UnitState):
+    reason: str
+    action = "repair the unreadable source; do not infer that the unit is idle"
+
+    @classmethod
+    def derive(cls, source: UnitObservation) -> Iterator[Self]:
+        # Validate the checkout reads before a higher-level label can hide a failure.
+        source.require_read()
+        return iter(())
+
+    def readable_units(self) -> tuple[ThreadIncarnation, ...]:
+        return ()
+
+    def merged_orders(self, comms: Comms, pulls: PullRequests) -> tuple[Todo, ...]:
+        # The card reports the failure; a merge read must not retry that source.
+        return ()
+
+    def detail(self) -> str:
+        return f"{super().detail()}; {self.reason}"
+
+
+@dataclass(frozen=True)
+class Blocked(NeedsCommanderAttention, UnitState):
+    goal: GoalSnapshot
+    action = "follow the blocked goal's declared response"
+
+    @classmethod
+    def derive(cls, source: UnitObservation) -> Iterator[Self]:
+        if not source.pending:
+            return
+        if source.goal.execution.unit_blocks():
+            yield cls(
+                source.thread.incarnation,
+                tuple(todo.id for todo in source.pending),
+                source.presence,
+                source,
+                source.goal,
+            )
+
+    def line(self) -> str:
+        return "\n".join(
+            block.unit_block_line(self.unit.name, self.orders)
+            for block in self.goal.execution.unit_blocks()
+        )
+
+
+@dataclass(frozen=True)
+class Detached(NeedsCommanderAttention, UnitState):
+    unread: int = field(compare=False)
+    action = "wake it or reassign the order"
+
+    @classmethod
+    def derive(cls, source: UnitObservation) -> Iterator[Self]:
+        if source.pending and not source.presence.receiving:
+            yield cls(
+                source.thread.incarnation,
+                source.orders,
+                source.presence,
+                source,
+                source.comms.bus.pending_count(source.thread.name),
+            )
+
+    def detail(self) -> str:
+        return f"{self.unread} unread; {super().detail()}"
+
+
+@dataclass(frozen=True)
+class AwaitingCommander(NeedsCommanderAttention, UnitState):
+    seconds: ClassVar[int] = 5 * 60
+    action = "answer the original message"
+
+    @classmethod
+    def derive(cls, source: UnitObservation) -> Iterator[Self]:
+        if source.due_questions:
+            yield cls(source.thread.incarnation, source.orders, source.presence, source)
+
+    def detail(self) -> str:
+        return ", ".join(
+            f"message {message.reference.seq}:{message.message_id}" for message in self.owed
+        )
+
+
+class Stuck(NeedsCommanderAttention, PendingWorkState, UnitState):
+    seconds: ClassVar[int] = 20 * 60
+    action = "inspect its last turn and restart only known-safe work"
+
+    @classmethod
+    def matches(cls, source: UnitObservation) -> bool:
+        calls = source.calls
+        latest = max(
+            (event.timestamp for turn in source.turns for event in turn.events),
+            default=source.since,
+        )
+        return bool(
+            not calls
+            and any(
+                todo.order.checked().first_call_deadline.before(source.now)
+                for todo in source.pending
+            )
+            or calls
+            and latest < source.now - cls.seconds
+        )
+
+
+class Spinning(NeedsCommanderAttention, PendingWorkState, UnitState):
+    action = "read the failed approach and change it before another rerun"
+
+    @classmethod
+    def matches(cls, source: UnitObservation) -> bool:
+        sends = sorted(
+            verdict.at.recorded_at("A send-back")
+            for todo in source.pending
+            for verdict in todo.verdicts
+            if isinstance(verdict, SendBackVerdict)
+        )
+        calls = source.calls
+        repeats = len(calls) >= 2 and calls[-1].digest == calls[-2].digest
+        if len(sends) < 2 and not repeats:
+            return False
+        commit_at = max(
+            (commit.at for origin in source.sources for commit in origin.commits()),
+            default=source.since,
+        )
+        return (
+            len(sends) >= 2
+            and sends[-2] > commit_at
+            or repeats
+            and source.turns[-1].start.since(commit_at)
+        )
+
+
+class StalledOnNothing(NeedsCommanderAttention, PendingWorkState, UnitState):
+    action = "resume its work or give it the missing next step"
+
+    @classmethod
+    def matches(cls, source: UnitObservation) -> bool:
+        return bool(
+            source.goal.execution.runnable()
+            and source.turns
+            and not source.comms.bus.pending_count(source.thread.name)
+        )
+
+
+class DeliveredUnclosed(NeedsCommanderAttention, ConditionalState, UnitState):
+    action = "install the merge or close with the working application evidence"
+
+    @classmethod
+    def matches(cls, source: UnitObservation) -> bool:
+        progress: dict[str, OrderProgress] = {
+            origin.todo.id: next(MergedProgress.found(origin), NoCommitProgress())
+            for origin in source.sources
+        }
+        return any(
+            todo.order.accepted or todo.order.waiting_on_install(todo, progress)
+            for todo in source.todos
+        )
+
+
+class Moving(ConditionalState, UnitState):
+    action = "let it continue"
+
+    @classmethod
+    def matches(cls, source: UnitObservation) -> bool:
+        return bool(source.todos)
+
+
+class Idle(ConditionalState, UnitState):
+    action = "give it the next useful order"
+
+    @classmethod
+    def matches(cls, source: UnitObservation) -> bool:
+        return not source.todos
diff --git a/tests/test_commander_forks.py b/tests/test_commander_forks.py
index 7b5c03aebc..92f8e9b257 100644
--- a/tests/test_commander_forks.py
+++ b/tests/test_commander_forks.py
@@ -58,7 +58,7 @@ def fork_bus(tmp_path):
         text='Repair the commander workflow', goal_id='request', order=RequestGoalOrder()))
     core = Repo.of(Path(__file__).parents[1])
     config = Commander.on_machine('lead', (source, core), comms.root, tmp_path)
-    config.root_path(comms.root).write_text(config.model_dump_json())
+    config.path_in(comms.root).write_text(config.model_dump_json())
     now = time.time()
     order = PendingProblemOrder(
         id='first-order', owner=source.scope().shape.parents[0], scope=source.scope(),
diff --git a/tests/test_goal_execution.py b/tests/test_goal_execution.py
index 5e9fe3b4d7..54c1ba3a4b 100644
--- a/tests/test_goal_execution.py
+++ b/tests/test_goal_execution.py
@@ -47,8 +47,11 @@ def execution_bus(copied_goal_bus):
 def test_execution_identity_survives_wire_and_change_detection(execution_bus, execution_type):
     goal = execution_bus.registry.require("worker").goal
     execution = execution_type(goal.id)
-    assert FieldCodec.decode(GoalExecution, FieldCodec.encode(execution)) == execution
+    decoded = FieldCodec.decode(GoalExecution, FieldCodec.encode(execution))
+    assert decoded == execution
+    assert decoded.runnable() == execution.runnable()
     current = GoalSnapshot(goal, RunnableExecution(goal.id))
+    assert current.execution.runnable()
     assert current.changed_by(GoalSnapshot(goal, execution)) == (current.execution != execution)
 
 
diff --git a/tests/test_holder_standing_rules.py b/tests/test_holder_standing_rules.py
index 5884abafcd..2ec4968014 100644
--- a/tests/test_holder_standing_rules.py
+++ b/tests/test_holder_standing_rules.py
@@ -17,10 +17,7 @@
 from agent_comms.commander.config import CommanderConfig
 from agent_comms.commander.order import Commander, OrderFile, SayVerb, SendVerb
 from agent_comms.commander.standing import (
-    Detached,
-    GoalCannotRun,
     HeartbeatSilent,
-    HolderCannotRun,
     UnitAskUnanswered,
 )
 from agent_comms.commander_heartbeat import CommanderHeartbeat
@@ -38,23 +35,27 @@
 from agent_comms.installer.backup import StoreBackup
 from agent_comms.launch_root import LaunchRoot
 from agent_comms.message_page import MessagePageRequest
-from agent_comms.messages import InfoKind, Message
+from agent_comms.messages import InfoKind, Message, QuestionKind
 from agent_comms.native_arguments import GivenBinary
 from agent_comms.order_runs import MechanicalOrderRun
 from agent_comms.read_basis import DMDisplayScope
 from agent_comms.recorded_time import RecordedEventTime
 from agent_comms.repository_name import RepositoryName
 from agent_comms.standing_rules import StandingRule
+from agent_comms.commander.questions import UnansweredUnitAsks
 from agent_comms.thread_identity import CoreWatcherIncarnation
 from agent_comms.thread_model import DeclaredModel
 from agent_comms.thread_status import StoppedThreadStatus
 from agent_comms.todos import (
+    ClaimOrder, OrderClaim, OrderMessage,
     PendingProblemOrder,
+    PullRequests,
     TodoConflict,
 )
 from conftest import declare_test_bus
 from native_backend_fixture import native_backend_fixture, wait_for
 from test_goal_execution import turn_on
+from agent_comms.ui_model.unit_state import AwaitingCommander, Blocked, Detached, UnitState
 
 
 @pytest.fixture
@@ -106,6 +107,36 @@ def test_stop_rejects_a_human_before_any_registry_change(comms, tmp_path):
     assert comms.registry.store.path.read_bytes() == before
 
 
+@pytest.fixture
+def claim_order(held_bus):
+    """Publish the real order and claim messages, using their retained identities."""
+    def claim(problem):
+        todo = held_bus.goals.todos.get(problem)
+        lead = held_bus.registry.require("lead").incarnation
+        worker = held_bus.registry.require("worker").incarnation
+        original = held_bus.messaging.send("lead", "worker", todo.order.instructions(todo)).message
+        todo = held_bus.goals.todos.apply(
+            todo.id, OrderMessage(todo.revision, lead, original.message_id))
+        claimed = OrderClaim.for_order(todo, worker, 1, "a" * 40,
+                                      "Review the published repair", original.reference)
+        claimed = replace(claimed, message=replace(
+            claimed.message, timestamp=time.time() - AwaitingCommander.seconds - 1))
+        ClaimOrder(todo.revision, worker, todo.held().generation, claimed).record(
+            held_bus.goals, todo.id)
+        return claimed
+    return claim
+
+
+def state_cards(comms, member):
+    return tuple(state for state in UnitState.current(comms, "lead", PullRequests())
+                 if isinstance(state, member))
+
+
+def unanswered(comms):
+    return UnitAskUnanswered.broken(
+        comms, "lead", observed=UnitState.current(comms, "lead", PullRequests()))
+
+
 def blocked_holder(comms):
     """Settle a real failed turn; never manufacture a stored backend block."""
     owner = comms.registry.require("worker")
@@ -127,7 +158,7 @@ def test_an_active_goal_with_an_unresolved_attempt_is_reported_never_retried(hel
     store.recover_unreserved_ready(goal.id, generation.number)
     store.reserve(goal.id, generation.number, ready_grant=store.ready_grant(goal.id, generation.number))
     reserved = store.require_generation(goal.id)
-    violation, = GoalCannotRun.violations(held_bus, "lead")
+    violation, = state_cards(held_bus, Blocked)
     assert violation.orders == ("P1", "P2")
     violation.enforce(held_bus, "lead")
     assert held_bus.registry.require("worker").goal.require() == goal
@@ -137,8 +168,8 @@ def test_an_active_goal_with_an_unresolved_attempt_is_reported_never_retried(hel
 
 def publish(comms, sender, target, timestamp, *, answers=()):
     return comms.bus.publisher.publish_ordinary(
-        Message(sender=sender, target=target, type=InfoKind, body="Not a question", timestamp=timestamp,
-                answers=answers), origin=AgentOrigin(),
+        Message(sender=sender, target=target, type=QuestionKind, body="Inspect the pending work",
+                timestamp=timestamp, answers=answers), origin=AgentOrigin(),
     )
 
 
@@ -150,14 +181,14 @@ def test_failed_holder_is_reported_without_reissuing_orders_or_replaying_unknown
     orders = held_bus.goals.todos.held_orders()
     inputs = InputDispositions(held_bus.root)
     unresolved = inputs.lookup("unresolved-input")
-    violation, = GoalCannotRun.violations(held_bus, "lead")
+    violation, = state_cards(held_bus, Blocked)
     assert violation.orders == ("P1", "P2")
     violation.enforce(held_bus, "lead")
     assert held_bus.registry.require("worker").goal == goal
     assert held_bus.goals.todos.held_orders() == orders
     assert store.require_generation(goal.id) == generation
     assert inputs.lookup("unresolved-input") == unresolved
-    assert GoalCannotRun.violations(held_bus, "lead") == (violation,)
+    assert state_cards(held_bus, Blocked) == (violation,)
 
 
 def test_an_order_to_a_blocked_goal_activates_it_and_keeps_other_orders(held_bus):
@@ -194,7 +225,7 @@ async def test_sigterm_group_preserves_active_goal_and_never_reissues_orders(
                 await Coordination.run_worker(partial(held_bus.owners.stop, "worker"))
                 assert held_bus.registry.require("worker").goal == goal
                 assert isinstance(held_bus.goals.goal_execution("worker"), RunnableExecution)
-                for violation in GoalCannotRun.violations(held_bus, "lead"):
+                for violation in state_cards(held_bus, Blocked):
                     violation.enforce(held_bus, "lead")
                 assert held_bus.goals.todos.held_orders() == orders
                 assert InputDispositions(held_bus.root).lookup("unresolved-input") == unresolved
@@ -223,7 +254,7 @@ def test_a_running_turn_is_not_a_stalled_holder(held_bus):
     lease = owner.thread.turn_lease
     assert lease is not None
     try:
-        assert GoalCannotRun.violations(held_bus, "lead") == ()
+        assert state_cards(held_bus, Blocked) == ()
     finally:
         held_bus.agents.finish_turn(lease)
 
@@ -231,21 +262,21 @@ def test_a_running_turn_is_not_a_stalled_holder(held_bus):
 def test_unanswered_uses_age_direct_target_and_later_answer_references(held_bus):
     now = time.time()
     publish(held_bus, "worker", "lead", now - 7200)  # before every held order
-    publish(held_bus, "worker", "lead", now - UnitAskUnanswered.seconds + 30)
+    publish(held_bus, "worker", "lead", now - AwaitingCommander.seconds + 30)
     publish(held_bus, "outsider", "lead", now - 1800)  # not a holder
     publish(held_bus, "worker", "outsider", now - 1800)  # not its commander
     question = publish(held_bus, "worker", "lead", now - 901)
     # An unrelated later answer and message text do not answer this reference.
     publish(held_bus, "lead", "worker", now - 800)
-    violation, = UnitAskUnanswered.violations(held_bus, "lead")
+    violation, = unanswered(held_bus)
     assert violation.message.reference == question.reference
     print(violation.line())
     held_bus.messaging.send("lead", "worker", "Responding", answers=(question.reference,))
-    assert UnitAskUnanswered.violations(held_bus, "lead") == ()
+    assert unanswered(held_bus) == ()
     another = publish(held_bus, "worker", "lead", now - 950)
     held_bus.messaging.send("lead", "outsider", "Answer in another conversation",
                             answers=(another.reference,))
-    assert UnitAskUnanswered.violations(held_bus, "lead") == ()
+    assert unanswered(held_bus) == ()
 
 
 @pytest.mark.parametrize('target,body', [
@@ -254,15 +285,15 @@ def test_unanswered_uses_age_direct_target_and_later_answer_references(held_bus)
 ])
 def test_commander_say_answers_an_unanswered_unit_message(held_bus, target, body, tmp_path):
     question = held_bus.bus.publisher.publish_ordinary(
-        Message(sender='worker', target=target, type=InfoKind, body=body,
-                timestamp=time.time() - UnitAskUnanswered.seconds - 1), origin=AgentOrigin(),
+        Message(sender='worker', target=target, type=QuestionKind, body=body,
+                timestamp=time.time() - AwaitingCommander.seconds - 1), origin=AgentOrigin(),
     )
     ctx = Commander.read_root(held_bus.root)
     reply = tmp_path / 'reply.txt'
     reply.write_text('Responding to the worker with the actual commander command')
     say = SayVerb(units="worker", body_file=str(reply), answers=(question.message_id,))
     say.apply(ctx)
-    assert UnitAskUnanswered.violations(held_bus, "lead") == ()
+    assert unanswered(held_bus) == ()
     # A name that is not open needs no answer: it is dropped, never a refusal of the verb.
     assert SayVerb(
         units="worker", body_file="unused", answers=("not-a-message",)
@@ -270,7 +301,7 @@ def test_commander_say_answers_an_unanswered_unit_message(held_bus, target, body
 
 
 def test_an_unanswered_unit_message_gates_send_and_say_until_answered(held_bus, tmp_path):
-    question = publish(held_bus, "worker", "lead", time.time() - UnitAskUnanswered.seconds - 1)
+    question = publish(held_bus, "worker", "lead", time.time() - AwaitingCommander.seconds - 1)
     ctx = Commander.read_root(held_bus.root)
     reply = tmp_path / "reply.txt"
     reply.write_text("Waiting on the merge queue, owner unit-goalblock")
@@ -279,7 +310,7 @@ def test_an_unanswered_unit_message_gates_send_and_say_until_answered(held_bus,
         with pytest.raises(SystemExit, match=question.message_id):
             verb.apply(ctx)
     SayVerb(units="worker", body_file=str(reply), answers=(question.message_id,)).apply(ctx)
-    assert UnitAskUnanswered.violations(held_bus, "lead") == ()
+    assert unanswered(held_bus) == ()
     # Naming it again after it was answered names nothing open, so it answers nothing.
     assert SendVerb(
         unit="worker", body_file="unused", answers=(question.message_id,)
@@ -287,18 +318,18 @@ def test_an_unanswered_unit_message_gates_send_and_say_until_answered(held_bus,
 
 
 def test_answering_one_owed_message_leaves_another_owed(held_bus, tmp_path):
-    first = publish(held_bus, "worker", "lead", time.time() - UnitAskUnanswered.seconds - 2)
-    second = publish(held_bus, "worker", "lead", time.time() - UnitAskUnanswered.seconds - 1)
+    first = publish(held_bus, "worker", "lead", time.time() - AwaitingCommander.seconds - 2)
+    second = publish(held_bus, "worker", "lead", time.time() - AwaitingCommander.seconds - 1)
     ctx = Commander.read_root(held_bus.root)
     reply = tmp_path / "reply.txt"
     reply.write_text("Waiting on the merge queue, owner unit-goalblock")
     with pytest.raises(SystemExit, match=second.message_id):
         SayVerb(units="worker", body_file=str(reply), answers=(first.message_id,)).apply(ctx)
-    assert {owed.message.message_id for owed in UnitAskUnanswered.violations(held_bus, "lead")} == {
+    assert {owed.message.message_id for owed in unanswered(held_bus)} == {
         first.message_id, second.message_id}
     SayVerb(units="worker", body_file=str(reply),
             answers=(first.message_id, second.message_id)).apply(ctx)
-    assert UnitAskUnanswered.violations(held_bus, "lead") == ()
+    assert unanswered(held_bus) == ()
 
 
 def test_an_overdue_owner_directive_gates_until_answered(held_bus, tmp_path):
@@ -319,8 +350,9 @@ def test_an_overdue_owner_directive_gates_until_answered(held_bus, tmp_path):
 
 @pytest.mark.parametrize('body', ['@outsider This question is yours', 'The worker broadcasts its result'])
 def test_channel_messages_without_a_commander_addressee_are_not_a_question(held_bus, body):
-    held_bus.messaging.send('worker', '#comms', body)
-    assert UnitAskUnanswered.asks(held_bus, 'lead', time.time()) == ()
+    held_bus.messaging.send('worker', '#comms', body, type=QuestionKind)
+    todos = StandingRule.todos(held_bus, 'lead')
+    assert UnansweredUnitAsks.for_orders(held_bus, 'lead', todos, time.time()) == ()
 
 
 def test_send_and_check_accept_a_deletion_check_on_fresh_main(
@@ -388,24 +420,22 @@ def cli(*arguments):
     print('Deletion expected 0, remote main reads 4: send and check accepted; stale dirty tree unchanged.')
 
 
-def test_detached_and_goal_blocks_are_both_in_standing_rule_scan(held_bus, installed_runtime):
-    blocked_holder(held_bus)
+def test_one_state_per_unit_even_when_blocked_and_detached(held_bus, installed_runtime):
     held_bus.registry.register(held_bus.registry.require("worker"), StoppedThreadStatus())
-    detached, = Detached.violations(held_bus, "lead")
-    assert detached.unit == "worker" and set(detached.orders) == {"P1", "P2"}
-    family = HolderCannotRun.violations(held_bus, "lead")
-    assert detached in family
-    blocked, = GoalCannotRun.violations(held_bus, "lead")
-    broken = StandingRule.broken(held_bus, "lead")
-    assert blocked in broken
-    assert all(rule in broken for rule in family)
+    detached, = state_cards(held_bus, Detached)
+    assert detached.unit.name == "worker" and set(detached.orders) == {"P1", "P2"}
+    blocked_holder(held_bus)
+    blocked, = state_cards(held_bus, Blocked)
+    states = UnitState.current(held_bus, "lead", PullRequests())
+    assert tuple(state for state in states if state.unit == blocked.unit) == (blocked,)
+    assert blocked not in StandingRule.broken(held_bus, "lead", observed=states)
 
 
 def test_startup_heartbeat_enforces_and_publishes_both_new_rule_lines(held_bus, installed_runtime):
     blocked_holder(held_bus)
     question = publish(held_bus, "worker", "lead", time.time() - 901)
     expected = tuple(rule.line() for rule in (
-        *GoalCannotRun.violations(held_bus, "lead"), *UnitAskUnanswered.violations(held_bus, "lead"),
+        *state_cards(held_bus, Blocked), *unanswered(held_bus),
     ))
     heartbeat = CommanderHeartbeat()
     commander = held_bus.registry.require("lead")
@@ -432,9 +462,10 @@ def test_every_pass_beats_and_a_stale_beat_is_heartbeat_silent(held_bus, monkeyp
     review = CommanderReview(frozenset())
     CommanderHeartbeat.register(held_bus.registry)
     assert HeartbeatSilent.violations(held_bus, "lead") == ()
+    states = UnitState.current(held_bus, "lead", PullRequests())
     with held_bus.bus.locked():
-        review.wake(held_bus, CommanderHeartbeat.sender, commander, ())
-        review.wake(held_bus, CommanderHeartbeat.sender, commander, ())
+        review.wake(held_bus, CommanderHeartbeat.sender, commander, (), states)
+        review.wake(held_bus, CommanderHeartbeat.sender, commander, (), states)
     page = held_bus.bus.log.read_page(MessagePageRequest(DMDisplayScope.capture(
         CommanderHeartbeat.sender, "lead", held_bus.registry.snapshot(),
     )))
diff --git a/tests/test_install.py b/tests/test_install.py
index 92c826f279..2f4cb84253 100644
--- a/tests/test_install.py
+++ b/tests/test_install.py
@@ -169,7 +169,7 @@ def test_an_install_on_a_bus_with_no_commander_writes_no_commander_config(privat
     home = _harnessless_home(private_directory, monkeypatch)
     result = InstallCliCommand(project=private_directory).apply(comms)
     assert result.config == NoConfigFile()
-    assert not CommanderConfig.root_path(root).exists()
+    assert not CommanderConfig.path_in(root).exists()
     assert not CommanderConfig.path(home).exists() and not CommanderConfig.path(home).is_symlink()
 
 
@@ -187,7 +187,7 @@ def test_an_install_refuses_flags_a_bus_without_a_commander_record_cannot_use(
     home = _harnessless_home(private_directory, monkeypatch)
     with pytest.raises(UpgradeRefused):
         InstallCliCommand(project=private_directory, **flags).apply(Comms(root))
-    assert not CommanderConfig.root_path(root).exists() and not CommanderConfig.path(home).is_symlink()
+    assert not CommanderConfig.path_in(root).exists() and not CommanderConfig.path(home).is_symlink()
 
 
 def test_an_install_without_flags_keeps_the_recorded_commander_and_rehomes_its_paths(
@@ -206,7 +206,7 @@ def test_an_install_without_flags_keeps_the_recorded_commander_and_rehomes_its_p
     result = InstallCliCommand(project=private_directory).apply(comms)
     expected = CommanderConfig.on_machine("lead", (), root, home, frozenset({"parked-unit"}))
     assert result.config == WrittenConfigFile(str(CommanderConfig.path(home)))
-    assert CommanderConfig.root_path(root).read_text() == expected.model_dump_json(indent=1) + "\n"
+    assert CommanderConfig.path_in(root).read_text() == expected.model_dump_json(indent=1) + "\n"
     assert CommanderConfig.read(CommanderConfig.path(home)) == expected
 
 
diff --git a/tests/test_proof_build.py b/tests/test_proof_build.py
index f6e8c01fbf..b8ee33e254 100644
--- a/tests/test_proof_build.py
+++ b/tests/test_proof_build.py
@@ -358,7 +358,7 @@ def test_install_due_compares_fetched_main_to_installed_build(
     before = Repo.git(checkout, "rev-parse", "HEAD")
     bus = Comms(tmp_path / "bus")
     config = CommanderConfig.on_machine("lead", (graph.repository,), bus.root, tmp_path / "home")
-    CommanderConfig.root_path(bus.root).write_text(config.model_dump_json())
+    CommanderConfig.path_in(bus.root).write_text(config.model_dump_json())
 
     due = InstallDue.violations(bus, "lead")
 
@@ -398,7 +398,7 @@ def test_install_waits_for_merges_to_settle_into_one_batch(
     Repo.git(checkout, "push", "origin", f"{graph.newer}:refs/heads/main")
     bus = Comms(tmp_path / "bus")
     config = CommanderConfig.on_machine("lead", (graph.repository,), bus.root, tmp_path / "home")
-    CommanderConfig.root_path(bus.root).write_text(config.model_dump_json())
+    CommanderConfig.path_in(bus.root).write_text(config.model_dump_json())
     merged = int(Repo.git(checkout, "log", "-1", "--format=%ct", graph.newer))
 
     monkeypatch.setattr(time, "time", lambda: merged + InstallDue.settle - 1)
diff --git a/tests/test_storage_violations.py b/tests/test_storage_violations.py
index 1dd225f9bb..687d7c0378 100644
--- a/tests/test_storage_violations.py
+++ b/tests/test_storage_violations.py
@@ -27,8 +27,9 @@
     StorageViolation,
     UnitStorage,
 )
-from agent_comms.todos import GoalRef, PendingProblemOrder, PresentCheckout
+from agent_comms.todos import GoalRef, PendingProblemOrder, PresentCheckout, PullRequests
 from agent_comms.order_runs import MechanicalOrderRun
+from agent_comms.ui_model.unit_state import UnitState
 
 
 @pytest.mark.parametrize(
@@ -129,7 +130,8 @@ def observed_read(_code: CodeType, _offset: int, value: UnitStorage) -> None:
         monitoring.set_local_events(
             monitoring.PROFILER_ID, UnitStorage.read.__func__.__code__, monitoring.events.PY_RETURN
         )
-        found = StorageViolation.broken(comms, "lead")
+        found = StorageViolation.broken(
+            comms, "lead", observed=UnitState.current(comms, "lead", PullRequests()))
     finally:
         monitoring.free_tool_id(monitoring.PROFILER_ID)
     assert {type(violation) for violation in found} == expected
diff --git a/tests/test_todos.py b/tests/test_todos.py
index 11a60931fa..60ac176649 100644
--- a/tests/test_todos.py
+++ b/tests/test_todos.py
@@ -236,7 +236,7 @@ def test_merged_order_keeps_its_checked_fields(tmp_path: Path, source_repository
 
 
 @pytest.fixture
-def pr_commander(tmp_path, monkeypatch, proof_build_checkouts):
+def pr_commander(tmp_path, monkeypatch, proof_build_checkouts, configured_repository):
     """One cited source, its real bare remote and installed history; only GitHub is synthetic."""
     graph = next(graph for graph in proof_build_checkouts
                  if graph.repository.github == CorePackage.repository)
@@ -267,6 +267,17 @@ def pr_commander(tmp_path, monkeypatch, proof_build_checkouts):
     rule = StorageRule(checkout_base=source.checkout.parent)
     (tmp_path / "storage.json").write_text(json.dumps(FieldCodec.encode(rule)))
     Registration(tmp_path / "registry.json").register(lead)
+    from agent_comms.plan_phases import PlanPhase
+
+    config = configured_repository(Comms(tmp_path), source)
+    declared = source.module_file(PlanPhase.__module__)
+    plans = config.module_repo(PlanPhase.__module__)
+    target = source.checkout / declared
+    target.parent.mkdir(parents=True, exist_ok=True)
+    target.write_bytes((plans.checkout / plans.module_file(PlanPhase.__module__)).read_bytes())
+    source.git(source.checkout, 'add', str(declared))
+    source.git(source.checkout, 'commit', '-qm', 'Declare the fixture plan owners')
+    source.git(source.checkout, 'push', 'origin', 'main')
     return Commander(commander=lead.name, root=tmp_path, state=tmp_path / "state",
                      repos=(source,), worktrees=source.checkout.parent,
                      scratch=tmp_path, disk=tmp_path)
diff --git a/tests/test_unit_state.py b/tests/test_unit_state.py
new file mode 100644
index 0000000000..24936ce871
--- /dev/null
+++ b/tests/test_unit_state.py
@@ -0,0 +1,253 @@
+"""Unit cards from real records, with unread failures isolated to their unit."""
+
+import json
+import time
+from dataclasses import replace
+from datetime import UTC, datetime
+from pathlib import Path
+
+import pytest
+
+from agent_comms.commander_heartbeat import CommanderHeartbeat
+from agent_comms.launch_runtime import LaunchRoot
+from agent_comms.message_page import MessagePageRequest
+from agent_comms.read_basis import DMDisplayScope
+from agent_comms.recorded_time import RecordedEventTime
+from agent_comms.thread_status import StoppedThreadStatus
+from agent_comms.thread_text import ForkedParent, SavedSessionFile
+from agent_comms.todos import DeliverTodo, OperationDelivery, PullRequests
+from agent_comms.ui_model.unit_state import (
+    AwaitingCommander,
+    Blocked,
+    DeliveredUnclosed,
+    Detached,
+    Idle,
+    Moving,
+    Spinning,
+    StalledOnNothing,
+    Stuck,
+    UnitState,
+    Unreadable,
+)
+from test_holder_standing_rules import blocked_holder, publish, state_cards
+
+pytest_plugins = ("test_holder_standing_rules",)
+
+
+def recorded_turn(comms, calls):
+    stamp = datetime.fromtimestamp(time.time(), UTC).isoformat()
+    rows = [
+        dict(type="session", id="state-source", version=3, timestamp=stamp, cwd=str(comms.root)),
+        dict(
+            type="message",
+            id="input",
+            timestamp=stamp,
+            message=dict(role="user", content="Repair the existing owner"),
+        ),
+    ]
+    for index in range(calls):
+        rows.append(
+            dict(
+                type="message",
+                id=f"call-{index}",
+                timestamp=stamp,
+                message=dict(
+                    role="assistant",
+                    content=[
+                        dict(
+                            type="toolCall",
+                            id=f"read-{index}",
+                            name="read",
+                            arguments=dict(path="owner.py"),
+                        )
+                    ],
+                    stopReason="toolUse",
+                ),
+            )
+        )
+        rows.append(
+            dict(
+                type="message",
+                id=f"result-{index}",
+                timestamp=stamp,
+                message=dict(
+                    role="toolResult",
+                    toolCallId=f"read-{index}",
+                    toolName="read",
+                    isError=False,
+                    content=[dict(type="text", text="contents")],
+                ),
+            )
+        )
+    rows.append(
+        dict(
+            type="message",
+            id="answer",
+            timestamp=stamp,
+            message=dict(
+                role="assistant",
+                content=[dict(type="text", text="Continue the repair")],
+                stopReason="stop",
+            ),
+        )
+    )
+    path = comms.root / "state-source.jsonl"
+    path.write_text("".join(json.dumps(row) + "\n" for row in rows))
+    path.chmod(0o600)
+    thread = comms.registry.require("worker")
+    comms.registry.register(replace(thread, session_file=SavedSessionFile(str(path))))
+
+
+def unread_checkout(comms, tmp_path):
+    checkout = tmp_path / "unread-worker"
+    checkout.mkdir()
+    (checkout / ".git").write_text("gitdir: ./removed-git-directory\n")
+    comms.registry.register(replace(comms.registry.require("worker"), worktree=str(checkout)))
+
+
+@pytest.mark.parametrize("member", UnitState.worst_first(), ids=lambda member: member.__name__)
+def test_unit_state_family_uses_real_records(held_bus, tmp_path, member):
+    comms = held_bus
+    unit = "worker"
+    with LaunchRoot.from_path(comms.root).selected():
+        if member is Unreadable:
+            unread_checkout(comms, tmp_path)
+        elif member is Idle:
+            unit = "outsider"
+        elif member is Blocked:
+            blocked_holder(comms)
+        elif member is Detached:
+            comms.registry.register(comms.registry.require(unit), StoppedThreadStatus())
+        elif member is AwaitingCommander:
+            publish(comms, unit, "lead", time.time() - AwaitingCommander.seconds - 1)
+        elif member is Stuck:
+            with comms.goals.todos.writing() as db:
+                for todo in comms.goals.todos.held_for_commander(
+                    comms.registry.require("lead").incarnation
+                ):
+                    replace(
+                        todo,
+                        order=replace(
+                            todo.order.checked(),
+                            first_call_deadline=RecordedEventTime(time.time() - 1),
+                        ),
+                    ).upsert(db)
+        elif member is Spinning:
+            recorded_turn(comms, 2)
+        elif member is StalledOnNothing:
+            recorded_turn(comms, 1)
+        elif member is DeliveredUnclosed:
+            for todo in comms.goals.todos.held_for_commander(
+                comms.registry.require("lead").incarnation
+            ):
+                comms.goals.todos.apply(
+                    todo.id,
+                    DeliverTodo(
+                        todo.revision,
+                        comms.registry.require("lead").incarnation,
+                        OperationDelivery(at=RecordedEventTime(time.time())),
+                        todo.order.id,
+                    ),
+                )
+        elif member is Moving:
+            pass  # Assigned work, before its first-call deadline.
+        else:
+            pytest.fail(f"New UnitState needs its real-record scenario: {member}")
+        states = UnitState.current(comms, "lead", PullRequests())
+        (state,) = (state for state in states if state.unit.name == unit)
+        assert isinstance(state, member)
+        assert state.unit == comms.registry.require(unit).incarnation
+        assert state.default_action and state.unit.name in state.line()
+        assert not any(
+            isinstance(state, sibling)
+            for sibling in UnitState.worst_first()
+            if sibling is not member
+        )
+        assert states == tuple(
+            state
+            for expected in UnitState.worst_first()
+            for state in states
+            if isinstance(state, expected)
+        )
+        print(state.line())
+
+
+def test_claim_is_a_commander_question_not_a_verified_delivery(held_bus, claim_order):
+    comms = held_bus
+    claim = claim_order("P1")
+    with LaunchRoot.from_path(comms.root).selected():
+        (state,) = state_cards(comms, AwaitingCommander)
+        (question,) = state.owed
+        assert question.message_id == claim.message.message_id
+        assert question.seq > claim.message.answers[0].seq
+        assert state_cards(comms, DeliveredUnclosed) == ()
+
+
+def test_stopped_child_without_lead_orders_remains_in_the_frame(held_bus):
+    child = replace(held_bus.registry.require("outsider"), parent=ForkedParent("lead"))
+    held_bus.registry.register(child, StoppedThreadStatus())
+    with LaunchRoot.from_path(held_bus.root).selected():
+        (state,) = (
+            state
+            for state in UnitState.current(held_bus, "lead", PullRequests())
+            if state.unit == child.incarnation
+        )
+        assert isinstance(state, Idle)
+        assert state.presence == StoppedThreadStatus()
+
+
+def test_malformed_native_source_is_unreadable_not_idle(held_bus):
+    recorded_turn(held_bus, 1)
+    path = held_bus.root / "state-source.jsonl"
+    rows = path.read_text().splitlines()
+    header = json.loads(rows[0])
+    del header["timestamp"]
+    path.write_text("\n".join((json.dumps(header), *rows[1:])) + "\n")
+    with LaunchRoot.from_path(held_bus.root).selected():
+        (state,) = state_cards(held_bus, Unreadable)
+        assert state.unit.name == "worker"
+        assert "timestamp" in state.reason
+
+
+@pytest.mark.parametrize(
+    "primary,broken_metadata",
+    ((False, True), (True, True), (False, False)),
+    ids=("home", "order", "nongit-home"),
+)
+def test_heartbeat_keeps_every_unit_when_one_checkout_is_unreadable(
+    held_bus, tmp_path, primary, broken_metadata, installed_runtime
+):
+    import os
+
+    from agent_comms.child_process import ProcessIdentity
+
+    if primary:
+        checkout = Path(held_bus.goals.todos.get("P1").order.checked().check.directory)
+        (checkout / ".git").rename(checkout / "saved-git")
+        (checkout / ".git").write_text("gitdir: ./removed-git-directory\n")
+    else:
+        unread_checkout(held_bus, tmp_path)
+    if not broken_metadata:
+        (held_bus.registry.require("worker").project_path / ".git").unlink()
+    lead = held_bus.registry.require("lead")
+    held_bus.registry.register(replace(lead, process_identity=ProcessIdentity.capture(os.getpid())))
+    heartbeat = CommanderHeartbeat()
+    with LaunchRoot.from_path(held_bus.root).selected():
+        observed = heartbeat.observe(held_bus)
+        cards = observed[0].states
+        assert isinstance(cards[0], Unreadable)
+        assert {card.unit.name for card in cards} >= {"worker", "outsider"}
+        heartbeat.pass_body(held_bus, observed)
+        page = held_bus.bus.log.read_page(
+            MessagePageRequest(
+                DMDisplayScope.capture(
+                    heartbeat.sender,
+                    "lead",
+                    held_bus.registry.snapshot(),
+                )
+            )
+        )
+        message = next(row for row in page.messages if row.sender == heartbeat.sender)
+        assert all(card.line() in message.body for card in cards)
+        assert message.body.index(cards[0].line()) < message.body.index(cards[-1].line())
+        print(message.body)
diff --git a/tests/test_unowned_checkout_work.py b/tests/test_unowned_checkout_work.py
index e84d4583eb..7f3d8f733b 100644
--- a/tests/test_unowned_checkout_work.py
+++ b/tests/test_unowned_checkout_work.py
@@ -17,13 +17,12 @@
 from agent_comms.commander.config import CommanderConfig
 from agent_comms.commander.standing import (
     CheckoutHeldWheelSpin,
-    CheckoutUnreadable,
     ReassignedWheelSpin,
     UnownedOrderWatch,
     UnownedWheelSpin,
     WaitingUnownedWheelSpin,
 )
-from agent_comms.commander_heartbeat import CommanderHeartbeat, CommanderReview
+from agent_comms.commander_heartbeat import CommanderHeartbeat, CommanderObservation, CommanderReview
 from agent_comms.comms import Comms
 from agent_comms.field_codec import FieldCodec
 from agent_comms.goal_actions import SetGoalAction
@@ -34,13 +33,12 @@
 from agent_comms.recorded_time import RecordedEventTime
 from agent_comms.repository_name import RepositoryName
 from agent_comms.storage_rules import StorageDenied, StorageRule, shell_storage_effects
-from agent_comms.standing_rules import StandingRule
 from agent_comms.thread_execution import ExternalThreadExecution
 from agent_comms.thread_identity import CoreWatcherIncarnation
-from agent_comms.todos import GoalRef, PendingProblemOrder, ReleaseTodo, StrikeReason
+from agent_comms.todos import GoalRef, PendingProblemOrder, PullRequests, ReleaseTodo, StrikeReason
 from agent_comms.transcript_events import AssistantTranscript
 from agent_comms.work_branches import UnreadableCheckoutRead, WorkBranches
-from order_scope_fixture import FIX_SCOPE
+from agent_comms.ui_model.unit_state import UnitState, Unreadable
 
 
 def git(directory: Path, *arguments: str) -> str:
@@ -104,7 +102,7 @@ def decide(command: str, cwd: Path) -> None:
 
 
 @pytest.fixture
-def checkout_bus(tmp_path: Path) -> Comms:
+def checkout_bus(tmp_path: Path, configured_repository, installed_runtime) -> Comms:
     home = tmp_path / 'declared-project'
     home.mkdir()
     git(home, 'init', '-q', '--initial-branch=main')
@@ -167,10 +165,7 @@ def checkout_bus(tmp_path: Path) -> Comms:
         turn_id='available', item_id='answer', events=(AssistantTranscript(
             'Ready for work', position='answer', timestamp=time.time()),))
     CommanderHeartbeat.register(c.registry)
-    CommanderConfig(
-        commander='lead', root=root, state=tmp_path / 'state', repos=(),
-        worktrees=rule.checkout_base, scratch=rule.scratch_base, disk=tmp_path,
-    ).write(tmp_path / 'config-home')
+    configured_repository(c)
     return c
 
 
@@ -197,11 +192,15 @@ def test_dirty_then_unpushed_then_pushed_linked_checkout(checkout_bus: Comms, tm
     with practice_copy(original.root, tmp_path / 'practice') as (root, _measurement):
         c = Comms(root)
         review = CommanderReview(frozenset())
+
+        def beat():
+            frame = CommanderObservation.read(c, c.registry.require('lead'))
+            review.wake(c, CommanderHeartbeat.sender, frame.thread, frame.rules, frame.states)
+
         with LaunchRoot.from_path(root).selected():
             (checkout / 'work.py').write_text('# not yet published\n')
             assert_retained(c)
-            review.wake(c, CommanderHeartbeat.sender, c.registry.require('lead'),
-                        StandingRule.broken(c, 'lead'))
+            beat()
             git(checkout, 'add', 'work.py')
             git(checkout, 'commit', '-q', '-m', 'Unpushed work')
             assert git(checkout, 'status', '--porcelain') == ''
@@ -214,20 +213,18 @@ def test_dirty_then_unpushed_then_pushed_linked_checkout(checkout_bus: Comms, tm
             assert isinstance(waiting[0], WaitingUnownedWheelSpin)
             UnownedOrderWatch().watch_pass(lambda: nullcontext(c))
             assert c.goals.todos.get('P2').holder is None
-            review.wake(c, CommanderHeartbeat.sender, c.registry.require('lead'),
-                        StandingRule.broken(c, 'lead'))
+            beat()
             config.write(tmp_path / 'copy-home')
             available = UnownedWheelSpin.violations(c, 'lead')
             assert len(available) == 1
             assert isinstance(available[0], ReassignedWheelSpin)
             assert available[0].unit == 'idle'
-            assert set(available).isdisjoint(review.violations)
+            assert set(available).isdisjoint(review.rules)
             UnownedOrderWatch().watch_pass(lambda: nullcontext(c))
             assigned = c.goals.todos.get('P2')
             assert assigned.holder == 'idle'
             assert c.registry.require('idle').goal.require().text
-            review.wake(c, CommanderHeartbeat.sender, c.registry.require('lead'),
-                        StandingRule.broken(c, 'lead'))
+            beat()
         drain_candidate_catchup(root)
         history = subprocess.run(
             (sys.executable, '-m', 'agent_comms.cli', '--root', str(root), 'history',
@@ -254,6 +251,8 @@ def test_unreadable_checkout_is_its_units_state_and_the_pass_carries_on(
             assert isinstance(read, UnreadableCheckoutRead) and read.holds_work
             UnownedOrderWatch().watch_pass(lambda: nullcontext(c))
             assert c.goals.todos.get('P2') == before
-            unreadable, = CheckoutUnreadable.violations(c, 'lead')
-            assert (unreadable.unit, unreadable.checkout) == ('worker', checkout)
+            unreadable, = (state for state in UnitState.current(c, 'lead', PullRequests())
+                          if isinstance(state, Unreadable))
+            assert unreadable.unit == c.registry.require('worker').incarnation
+            assert str(checkout) in unreadable.reason
             assert 'not a git repository' in unreadable.line()