Share commander HUD columns, grouped heartbeat and retained alarms

pyproject.toml

@@ -30,6 +30,7 @@ dependencies = [
     "nominal-refactor-advisor @ git+https://github.com/OpenHCSDev/NominalRefactorAdvisor@main",
     "packaging>=24",
     "pydantic>=2.13",
+    "python-introspect @ git+https://github.com/OpenHCSDev/python-introspect@main",
 ]
 
 [project.optional-dependencies]

src/agent_comms/activity.py

@@ -133,7 +133,7 @@ def from_error(cls, owner: OwnerIdentity, error: Exception) -> Self:
     def summary(self) -> str:
         return f"{self.headline} · {self.error_type}: {self.reason} · {self.recovery}"
 
-    @projected(view="artifact", name="kind")
+    @projected(view=ArtifactExclude, name="kind")
     def artifact_kind(self) -> str:
         return FieldCodec.encode(type(self))
 

src/agent_comms/child_process.py

@@ -43,7 +43,7 @@
 
 from .declared_family import DeclaredFamily
 from .errors import RelationViolationError
-from .field_codec import FieldCodec, FieldDirective, Id128, NullMember, RecordMember, SubclassUnion, WireRequired
+from .field_codec import DisclosedRecord, FieldCodec, FieldDirective, Id128, NullMember, RecordMember, SubclassUnion, WireRequired
 from .launch_runtime import INHERITED_ENVIRONMENT, LaunchRuntime, RestoredRuntime
 from .native_build_inputs import (
     AsyncStdioOptions, ChildStdio, ConfiguredChildStdio, DetachedChildStdio, GitAcquisitionCancelled,
@@ -1391,7 +1391,7 @@ class OrderCheckFailedError(ValueError):
     """A completed order check did not prove its expected result."""
 
 
-class CheckPrinted(DeclaredFamily, affix="Printed"):
+class CheckPrinted(DisclosedRecord, DeclaredFamily, affix="Printed"):
     """The count printed by a child, or the absence of an integer."""
 
     @staticmethod

src/agent_comms/cli_bridges/claude.py

@@ -5,11 +5,9 @@
 from contextlib import aclosing
 from functools import partial
 from pathlib import Path
-from typing import TYPE_CHECKING, AsyncIterator, BinaryIO
+from typing import AsyncIterator, BinaryIO
 
 from agent_comms.bus import Bus
-from agent_comms.harness_delivery_states import Outcome
-from agent_comms.messages import Message
 from agent_comms.private_path import FileRevision
 from agent_comms.runtime_updates import RuntimeUpdate
 from agent_comms.watched import FileSource
@@ -21,19 +19,14 @@
 from .claude_transcript import ClaudeJsonLines, RunningClaude
 from .shared import Enrol, LineMonitorBridge, Options, Watch, run_bridge
 
-if TYPE_CHECKING:
-    from agent_comms.task_attachments.commander_pings import CommanderUpdate
-
-
 class ClaudeBridge(ClaudeJsonLines, ClaudeHistory, LineMonitorBridge):
-    """Follow this Claude session's file without submitting an input.
+    """Follow the Claude session and deliver the original waking HUD through its native frame.
 
-    Heartbeat packets reach the session through the board its tool-call hook prints, which
-    this bridge derives from Core whenever the hook asks, not through the Monitor: the harness
-    frames Monitor events as routine background, so a line said once there scrolls away. The
-    Monitor gets the packet itself too, since an idle session runs no tool calls. It is the
-    same message the delivery attempt was begun for: a rewritten body hashes to another
-    message_id, which no attempt was stored under."""
+    Packet-owned wake behavior is shared with every bridge; tool-call board reads still derive
+    their current undisposed requests from Core rather than keeping another copy here.
+    The native frame receives the original message: rewriting its body changes the identity
+    the delivery attempt was stored under.
+    """
 
     incoming_frame = ClaudeMonitorFrame
     window_bytes = ClaudeMonitorFrame.BATCH_CHARS * 2
@@ -49,13 +42,6 @@ async def watch_operation(self, operation: Watch) -> None:
         await asyncio.to_thread(state.rearm, partial(Connect().begin, self))
         await super().watch_operation(operation)
 
-    async def deliver_board(self, update: CommanderUpdate, message: Message) -> type[Outcome]:
-        """The packet is on the bus, which is the board; only the session needs waking."""
-        return await update.wake(self, message)
-
-    async def wake(self, message: Message) -> type[Outcome]:
-        return await self.deliver(message)
-
     def history_start(self, stream: BinaryIO, boundary: int) -> int:
         """Retain the recent window's owning occurrence, never an arbitrary mid-turn prefix."""
         start = max(0, boundary - self.RECENT_BYTES)

src/agent_comms/cli_bridges/shared.py

@@ -1134,7 +1134,11 @@ async def prepare_delivery(self) -> None:
         """A transport may prepare its harness after the operation bound its reader."""
 
     async def deliver_board(self, update: CommanderUpdate, message: messages.Message) -> type[Outcome]:
-        """A harness with no tool-call route takes a heartbeat packet like any message."""
+        """Every transport honors the packet's wake decision; passive display is not input."""
+        return await update.wake(self, message)
+
+    async def wake(self, message: messages.Message) -> type[Outcome]:
+        """The packet already owns its HUD text; every harness receives that original message."""
         return await self.deliver(message)
 
     async def accept(self, message: messages.Message) -> None:

src/agent_comms/cli_options.py

@@ -7,10 +7,10 @@
 from collections.abc import Sequence
 from typing import ClassVar, Self
 
-from .field_codec import FieldCodec
+from .field_codec import DisclosedRecord, FieldCodec
 
 
-class JsonFlagRecord:
+class JsonFlagRecord(DisclosedRecord):
     """A record supplies its JSON-value flag; FieldCodec owns its representation."""
 
     __slots__ = ()

src/agent_comms/commander/publication_rules.py

@@ -1,3 +1,3 @@
 """Held work must be published, and a linked pull request must contain a change."""
 
 from __future__ import annotations
@@ -43,7 +43,7 @@
     def line(self) -> str:
         return (f"{self.holder.name} has {self.unpublished.commits} unpushed commits on"
                 f" {self.problem} ({self.unpublished.head[:9]} in {self.worktree}, upstream"
-                f" {self.unpublished.upstream or 'not configured'}). waiting for the disk to die?"
+                f" {self.unpublished.upstream or 'not configured'}). these commits exist only locally"
                 " -> push them")
 
 
@@ -62,7 +62,7 @@
         for held in branches:
             reader = held.repository.branches
             todo = held.todo
-            for pull in held.source.pulls.open_for(todo):
+            for pull in held.source.frame.pulls.open_for(todo):
                 for compared in pull.empty_changes(reader):
                     found.append(cls(held.problem, todo.order.checked().id, held.holder,
                                      PullRequestLink(todo.repo, pull.number),

src/agent_comms/commander/standing.py

@@ -1,3 +1,3 @@
 """Standing rules read from the commander's own files, members of Core's StandingRule family."""
 
 from __future__ import annotations
@@ -178,10 +178,10 @@
         """Due is a disposition: the commander's `order install` launches."""
 
     def line(self) -> str:
-        oldest = int((time.time() - min(merge.at for merge in self.merges)) // 60)
+        oldest = RecordedEventTime(min(merge.at for merge in self.merges)).stamp()
         landing = ", ".join(f"#{row.number} {row.branch}" for row in self.landing) or "none"
         return (f"{self.headline} {len(self.merges)} merges sitting on {self.repo.to_wire()} main {self.main[:12]},"
-                f" installed is {self.installed[:12]}, oldest {oldest} min"
+                f" installed is {self.installed[:12]}, oldest merge at {oldest}"
                 f" ({'; '.join(merge.line for merge in self.merges)}; landing: {landing})."
                 f" {self.question} -> {self.action}")
 
@@ -331,9 +331,8 @@
         )
 
     def line(self) -> str:
-        hours = (time.time() - self.since) / 3600
         return (
-            f"UNSTALL {self.phase.plan.declared_name}/{self.phase.declared_name} held {hours:.1f}h"
+            f"UNSTALL {self.phase.plan.declared_name}/{self.phase.declared_name} held since {RecordedEventTime(self.since).stamp()}"
             f" and still not on main ({self.phase.plan.document}). what's actually holding it?"
             " -> find it and change that this wake: integrate steps that only land together,"
             " take it over, or reassign it"
@@ -385,8 +384,7 @@
                 f"answer it in a message to {self.unit}")
 
     def line(self) -> str:
-        age = max(0, int((time.time() - self.message.timestamp) // 60))
-        return (f"ANSWER {self.unit} asked {self.cited} {age} min ago. why is it still waiting on you?"
+        return (f"ANSWER {self.unit} asked {self.cited} at {RecordedEventTime(self.message.timestamp).stamp()}. why is it still waiting on you?"
                 f" -> {self.clearing_action()}")
 
 

src/agent_comms/commander/vibe_check.py

@@ -8,6 +8,8 @@
 
 from __future__ import annotations
 
+from agent_comms.field_codec import WireExclude
+
 import time
 from collections.abc import Mapping, Sequence
 from abc import abstractmethod
@@ -25,6 +27,7 @@
     OwnerInvocable,
 )
 from agent_comms.messages import Message
+from agent_comms.recorded_time import RecordedEventTime
 from agent_comms.registry_document import RegistrySnapshot
 from agent_comms.task_attachments.owner_notices import ChokeholdNotice, EscalationNotice, VibeCheckNotice
 from agent_comms.standing_rules import (
@@ -93,11 +96,11 @@ def from_orders(cls, orders: Sequence[Todo], timers_up: int, unjudged: int) -> S
         """One load projection shared by the heartbeat's HUD and standing rules."""
         return cls(sum(todo.pending for todo in orders), timers_up, unjudged)
 
-    @projected(view="wire")
+    @projected(view=WireExclude)
     def total(self) -> float:
         return self.owed + 2 * self.timers_up + self.unjudged
 
-    @projected(view="wire")
+    @projected(view=WireExclude)
     def overloaded(self) -> bool:
         return self.total >= self.ceiling
 
@@ -173,8 +176,8 @@ def due(self, now: float) -> bool:
         return self.message.timestamp <= now - self.seconds
 
     def line(self) -> str:
-        age = max(0, int((time.time() - self.message.timestamp) // 60))
-        return f"vibe check {self.cited} is {age} min old. still ignoring it? -> answer it"
+        return (f"vibe check {self.cited} from {RecordedEventTime(self.message.timestamp).stamp()}."
+                " still ignoring it? -> answer it")
 
 
 class Indictment:

src/agent_comms/commander_heartbeat.py

@@ -8,9 +8,11 @@
 from dataclasses import dataclass
 from typing import TYPE_CHECKING
 
-from agent_comms.commander_hud import CommanderFrame, CommanderHud
+from agent_comms.commander_hud import CodexUsage, CommanderFrame, CommanderHud, HudUnavailableGroup
 from agent_comms.child_process import ReadWork
-from agent_comms.task_attachments.commander_pings import CommanderUpdate, RulePing, UnitPing
+from agent_comms.task_attachments.commander_pings import (
+    AlarmGroup, CommanderUpdate, PublishedFacts, RulePing, UnitPing,
+)
 from agent_comms.commander_service import CommanderService
 from agent_comms.heartbeat_frame import CommanderInputs, HeartbeatFrame
 from agent_comms.heartbeat_subjects import CommanderHistory
@@ -20,7 +22,7 @@
 from agent_comms.standing_rules import ServicePassFailing, StandingRule
 from agent_comms.threads import Thread
 from agent_comms.todos import ProgressRead, Todo
-from agent_comms.ui_model.unit_state import AwaitingCommander, UnitState
+from agent_comms.ui_model.unit_state import UnitState
 
 if TYPE_CHECKING:
     from agent_comms.comms import Comms
@@ -71,13 +73,18 @@ def failed(self, opened_wire: Callable[[], AbstractContextManager[Comms]],
             failure = ServicePassFailing(watch=name, since=retained.since, error=retained.error)
             now = time.time()
             snapshot = comms.registry.snapshot()
+            actions = CommanderHistory.read(comms, self.actors(snapshot))
             for thread in (recipient for lead in self.boards(comms.registry, comms.goals)
                            for recipient in self.recipients(snapshot, lead)):
                 for update in CommanderUpdate.report_failure(
                     thread.incarnation, failure,
                     CommanderUpdate.latest(comms, thread.incarnation), now=now,
                 ):
-                    update.send(comms, now, ())
+                    sections = AlarmGroup.sections((update,), update.open_pings(actions))
+                    update.send(comms, now, (
+                        HudUnavailableGroup.heading(),
+                        *CommanderHud.alarm_text(sections, now),
+                    ))
 
     def prepare(self, comms: Comms) -> Mapping[Thread, CommanderInputs]:
         return {thread: CommanderInputs.prepare(comms, thread.name, ProgressRead(), self)
@@ -90,47 +97,48 @@ def observe(self, comms: Comms, prepared: Mapping[Thread, CommanderInputs]) -> t
         )
 
     def pass_body(self, comms: Comms, observed: tuple[CommanderObservation, ...]) -> None:
-        """The committed packet survives connector replacement; local memory owns nothing.
-
-        One board is read per registered commander; every commander it serves receives that
-        board under its own incarnation and completes its own pass."""
+        """Read one shared board; each recipient keeps its own committed publication evidence."""
+        published = {}
         completed = []
-        huds = []
         now = time.time()
         try:
             for item in observed:
                 comms.goals.todos.record_merges(item.merges(comms))
                 recipients = self.recipients(comms.registry.snapshot(), item.thread)
-                held = {recipient: CommanderUpdate.latest(comms, recipient.incarnation)
+                held = {recipient.incarnation: CommanderUpdate.latest(comms, recipient.incarnation)
                         for recipient in recipients}
-                previous = held[item.thread]
+                previous = held[item.thread.incarnation]
                 known = frozenset(ping for packet in previous for ping in packet.handled)
                 rejected = StandingRule.enforce_all(
                     comms, item.thread.name,
                     (*(rule for rule in item.rules if RulePing.capture(rule) not in known),
                      *(state for state in item.states
                        if state.attention() and UnitPing.capture(state) not in known)),
                 )
-                # Board disposition reads CommanderHistory after enforcement.
-                hud = CommanderHud.of(comms, item.thread.name, item.frame, item.states)
+                usage = CodexUsage.read()
+                # Include acknowledgements committed during forge/provider reads and enforcement.
                 history = CommanderHistory.read(comms, self.actors(comms.registry.snapshot()))
                 states = tuple(state for state in item.states if state not in rejected)
                 rules = tuple(rule for rule in item.rules if rule not in rejected)
                 not_applied = tuple(rule.rejected_line(error) for rule, error in rejected.items())
-                for recipient, latest in held.items():
-                    for update in CommanderUpdate.derive(
-                        recipient.incarnation, states, rules, latest, now=now,
-                        undisposed=tuple(
-                            ping for packet in latest
-                            for ping in packet.undisposed(history, now - AwaitingCommander.seconds)),
-                        not_applied=not_applied,
-                    ):
-                        update.send(comms, now, hud.brief())
-                    completed.append(recipient.incarnation)
-                huds.append(hud)
+                deliveries = tuple(CommanderUpdate.derive(
+                    recipient, states, rules, latest, now=now, history=history,
+                    not_applied=not_applied,
+                ) for recipient, latest in held.items())
+                board = CommanderHud.of(comms, item.thread.name, item.frame, item.states,
+                                        delivery=deliveries[0], usage=usage)
+                for delivery in deliveries:
+                    hud = board.for_delivery(delivery)
+                    recipient = delivery.packet.commander
+                    published[recipient] = hud
+                    facts = PublishedFacts.of(hud.display_facts)
+                    latest = held[recipient]
+                    if delivery.packet.should_publish(facts, latest):
+                        delivery.packet.send(comms, now, hud.hud_text(now), publication=facts)
+                    completed.append(recipient)
         finally:
-            # A later commander's failure does not erase passes already completed.
-            CommanderFrame.replace_all(comms.root, tuple(huds))
+            # A later commander's failure does not erase frames/passes already completed.
+            CommanderFrame.replace_all(comms.root, tuple(published.values()))
             ServicePassStore.at(comms.root).heartbeat_passed(
                 completed,
                 (thread.incarnation for thread in comms.registry.all_threads().values()),