Integrate P738 P739 P707 P742 P728 P743 (#2061 #2066 #1952 #2080 #1988 #2085)

src/agent_comms/activity.py

@@ -103,9 +103,11 @@ class DrainDiagnostic(DeclaredFamily, affix="Diagnostic"):
     @abstractmethod
     def recovery(self) -> str: ...
 
-    @property
-    def headline(self) -> str:
-        return "Inbox unavailable"
+    headline: ClassVar[str] = "Inbox unavailable"
+
+    @classmethod
+    def from_error(cls, owner: OwnerIdentity, error: Exception) -> Self:
+        return cls(owner, type(error).__name__, str(error), recorded_at=time.time())
 
     @property
     def summary(self) -> str:
@@ -125,12 +127,14 @@ def recovery(self) -> str:
         return "Waiting for recovery"
 
 
+class FailedWakeDiagnostic(UnavailableDrainDiagnostic):
+    headline = "Wake failed"
+
+
 class NativeInputDrainDiagnostic(DrainDiagnostic):
     """A native request failure is not evidence that the inbox cannot be read."""
 
-    @property
-    def headline(self) -> str:
-        return "Native request failed"
+    headline = "Native request failed"
 
     @property
     def recovery(self) -> str:

src/agent_comms/bus_publication.py

@@ -189,10 +189,11 @@ def deliveries(self) -> tuple[CommittedDelivery, ...]:
     def require_claim(self, metadata) -> None:
         """Factory already bound root/audience; deliveries need no retained-public exception."""
 
-    def checkpoint_rows(self, offset: int, length: int, previous_sequence: int):
+    def checkpoint_rows(self, offset: int, length: int, previous_sequence: int,
+                        claimed_after_seq: int):
         from .private_bus_checkpoint import Addressed, DeliverySources, Replies, ResponseKeys
 
-        yield from super().checkpoint_rows(offset, length, previous_sequence)
+        yield from super().checkpoint_rows(offset, length, previous_sequence, claimed_after_seq)
         if self.acknowledgement is not None:
             yield ResponseKeys(self.acknowledgement.publication_key)
         yield DeliverySources(self.message.seq, self.message.message_id,

src/agent_comms/cli_commands.py

@@ -1237,7 +1237,7 @@ def apply(self, ctx: Comms) -> CliResult:
         home = Path.home()
         project = self.project.resolve()
         config = CommanderConfig(
-            commander=self.commander, root=ctx.root, cli=Harness.bin("agent-comms"),
+            commander=self.commander, root=ctx.root,
             state=home / ".local/state/agent-comms/commander",
             board=project / "docs/agents/fleet-board.md", narrative=project / "docs/agents/fleet-narrative.md",
             repos=tuple(Repo.of(repo) for repo in self.repos or (project,)),

src/agent_comms/commander/config.py

@@ -35,7 +35,6 @@ class CommanderConfig(CommanderRole, BaseModel):
 
     commander: str
     root: Path
-    cli: Path
     state: Path
     board: Path
     narrative: Path

src/agent_comms/commander/order.py

@@ -105,19 +105,6 @@ def struck(todo, unit):
     return len(todo.live_strikes(unit))
 
 
-def hhmm(event):
-    return time.strftime('%H:%M', time.localtime(event.at))
-
-
-def deadline(due):
-    """HH:MM local time: the first such minute at or after now."""
-    hour, minute = map(int, due.split(':'))
-    now = time.time()
-    day = time.localtime(now)
-    at = time.mktime((day.tm_year, day.tm_mon, day.tm_mday, hour, minute, 0, 0, 0, -1))
-    return RecordedEventTime(at + 86400 if at < now - 60 else at)
-
-
 # Eliminate forwarding with inheritance and multiple parents; decompose the hierarchy if it is not sound.
 # A forwarder restates an implementation pair (paper 1 Fig 3: fields + forwarding leave routes to maintain), so
 # nothing the commander approves, delivers or orders may add or keep one. The cure is the hierarchy, never a
@@ -404,14 +391,17 @@ def send(self, unit, due, document: OrderFile, clean=False, reassign=None, new=N
         # A miss already struck on the order this replaces is one miss:
         # the send-back records no second strike.
         struck_already = replacing and listed[group[0]].struck_on_order(unit)
-        common = dict(order=order, request=order.request(self.comms.registry.require(self.commander), tuple(group)), deadline=deadline(due), commander=self.commander)
+        request = order.request(self.comms.registry.require(self.commander), tuple(group))
+        due_at = RecordedEventTime(time.time()).at_clock(due)
         if new:
             self.call(CommsTodoCreateTool(owner=unit, order=CreateProblemGoalAction(
-                problem=new[0], repo=Repo.of(Path(check.directory)).github, text=new[1], **common)))
+                problem=new[0], repo=Repo.of(Path(check.directory)).github, text=new[1],
+                order=order, request=request, deadline=due_at, commander=self.commander)))
         for pid in [] if new else group:
             self.call(CommsOrderTool(owner=unit, order=OrderGoalAction(
                 problem=pid, expected_revision=self.comms.goals.todos.get(pid).revision,
-                assigner=self.incarnation, traced=trace, **common)))
+                assigner=self.incarnation, traced=trace,
+                order=order, request=request, deadline=due_at, commander=self.commander)))
         message += "\n" + "\n".join(order.pull_request_instruction(pid) for pid in group)
         sent = self.post(VerdictAudience((unit,)), message, answers)
         # The message is each order's recorded link; a verdict citing it lands on that order.
@@ -485,18 +475,15 @@ def write_board(self):
         lines = ['# Fleet board', '',
                  "GENERATED by agent-comms-commander order from Core's Todo store. Do not edit by hand:",
                  '`order problem FILE.json` adds a row, `order send` orders it, `order close` or `order withdraw` removes it.', '',
-                 '| Problem | Owner file | Owner | Order | Due | Progress | Delivered '
+                 '| Problem | Owner file | Owner | Order | Due | Check | Progress | Order state '
                  '| Strikes | Owner last acted |',
-                 '|---|---|---|---|---|---|---|---|---|']
+                 '|---|---|---|---|---|---|---|---|---|---|']
         listed, read = self.todos(), self.progress()
         activity = {unit: self.last_activity(unit) for unit in {t.holder for t in listed if t.holder}}
         for t in sorted(listed, key=lambda t: int(t.id[1:]) if t.id[1:].isdigit() else 0):
-            unit, last = t.holder, t.released[-1:]
-            owner = unit or f'unassigned: released from {last[0].previous.owner}: {last[0].reason.text}'
-            lines.append(f'| {t.id}: {t.text} | {t.order.owner.to_wire()} | {owner} | {t.order.id} | '
-                         f'{hhmm(t.deadline)} | {read[t.id].line if t.id in read else "-"} | '
-                         f'{"no" if t.pending else "yes, install pending"} | '
-                         f'{t.strike_count} | {activity[unit].clock() if unit else "-"} |')
+            lines.append(f'| {t.id}: {t.text.replace("|", "&#124;")} | {t.order.board_cells(t)} | '
+                         f'{read[t.id].line if t.id in read else "-"} | {t.order.line} | '
+                         f'{t.strike_count} | {activity[t.holder].clock() if t.holder else "-"} |')
         history = self.comms.goals.todos.history_for_commander(self.incarnation)
         lines += ['', '## Withdrawn orders', '',
                   *(f'- {line}' for t in history for line in t.order.withdrawals(t))]
@@ -619,7 +606,7 @@ def first_calls(self, held, activity):
             self.call(CommsChangeProblemTool(problem=todo.id, operation=RecordFirstCallTodo(
                 todo.revision, self.incarnation, event)))
             if not acted and self.comms.goals.todos.get(todo.id).revision != todo.revision:
-                missing.append(f'{unit} (order {order.id}, sent {hhmm(order.at)})')
+                missing.append(f'{unit} (order {order.id}, sent {order.at.clock()})')
         return missing
 
     def hunt(self, where, why, sent):
@@ -663,7 +650,7 @@ def deadline_lines(self, pending: list[Todo], read: dict[str, OrderProgress],
             if not t.overdue(now):
                 continue
             answer = self.reply(t.holder, ordered)
-            due = f'due {hhmm(t.deadline)}, order {t.order.id}'
+            due = f'due {t.deadline.clock()}, order {t.order.id}'
             late.append(f'{t.holder} ({due}, {work.line}{answer})')
         alarm = f'OVERDUE: {", ".join(late)} -> push or merge unfinished work now'
         return [alarm] * bool(late)
@@ -704,7 +691,7 @@ def overdue(self):
     def unstrike(self, task: str, why: str, at: str | None = None) -> None:
         """A strike recorded in error is voided in Core, then narrated and posted to #comms."""
         todo = self.comms.goals.todos.get(task)
-        live = [s for s in todo.live_strikes() if at in (None, hhmm(s.at))]
+        live = [s for s in todo.live_strikes() if at in (None, s.at.clock())]
         if not live:
             sys.exit(f'Refused: {task} has no live strike' + (f' recorded at {at}.' if at else '.'))
         voided = live[-1]
@@ -714,7 +701,7 @@ def unstrike(self, task: str, why: str, at: str | None = None) -> None:
         unit = voided.previous.owner
         n = struck(self.comms.goals.todos.get(task), unit)
         line = (f'{time.strftime("%Y-%m-%d %H:%M")} {unit} (strike voided on {task}, '
-                f'recorded {hhmm(voided.at)}; {n} strike(s) remain): {why}')
+                f'recorded {voided.at.clock()}; {n} strike(s) remain): {why}')
         self.on_record(line, 'Void', VerdictAudience.of(unit, todo.goal))
         print(f'{unit}: {n} strike(s) on {task}')
 
@@ -726,13 +713,13 @@ def due(self, pid: str, at: str, why: str) -> None:
         if len(why.split()) < 8:
             sys.exit('Refused: --why says why the deadline moves (8+ words)')
         refuse(why, *OUTGOING)  # posted to #comms, so refused before Core records the deadline
-        moved = deadline(at)
+        moved = RecordedEventTime(time.time()).at_clock(at)
         self.call(CommsChangeProblemTool(problem=pid, operation=AssignTodo(
             todo.revision, todo.assignment, moved, todo.order, todo.goal)))
         self.on_record(f'{time.strftime("%Y-%m-%d %H:%M")} {todo.holder} ({pid} order {todo.order.id} due '
-                  f'{hhmm(todo.deadline)} moved to {hhmm(moved)}): {why}', 'Deadline',
+                  f'{todo.deadline.clock()} moved to {moved.clock()}): {why}', 'Deadline',
                   VerdictAudience.of(todo.held().owner, todo.goal))
-        print(f'{pid}: due {hhmm(moved)}')
+        print(f'{pid}: due {moved.clock()}')
 
     @boundary_decoder
     def withdraw(self, pid: str, why: str, superseded_by: str | None = None) -> None:
@@ -924,13 +911,14 @@ def run(self, ctx: Commander) -> None:
 
 
 @dataclass(frozen=True, kw_only=True)
-class InstallVerb(RuledVerb):
+class InstallVerb(OrderVerb, RuledVerb):
     help = "plan the fleet's next install; merged and delivered orders wait on it"
     at: str = option("at", help="HH:MM the install is due")
     installer: str = option("installer", help="the unit that runs the install")
 
     def run(self, ctx: Commander) -> None:
-        PlannedInstallDeadline(deadline(self.at), self.installer).plan(ctx.comms.root)
+        planned = RecordedEventTime(time.time()).at_clock(self.at)
+        PlannedInstallDeadline(planned, self.installer).plan(ctx.comms.root)
         print(f'install due {self.at}, run by {self.installer}')
 
 

src/agent_comms/commander/request.py

@@ -16,7 +16,7 @@
 from ..goal_states import ActiveGoal
 from ..goals import Goal, RequestGoalOrder
 from ..todos import Todo
-from .order import Commander, CommanderVerb, RuledVerb, hhmm
+from .order import Commander, CommanderVerb, RuledVerb
 
 
 def open_requests(ctx: Commander) -> tuple[tuple[Goal, tuple[Todo, ...]], ...]:
@@ -64,7 +64,7 @@ def run(self, ctx: Commander) -> None:
         for goal, served in requests:
             print(f"{goal.id} {goal.text}")
             for todo in served:
-                print(f"  {todo.id} due {hhmm(todo.deadline)}: {todo.text}")
+                print(f"  {todo.id} due {todo.deadline.clock()}: {todo.text}")
 
 
 def main(a: list[str], ctx: Commander) -> None:

src/agent_comms/commander/rules/caution-must-pay.md

@@ -0,0 +1,9 @@
+---
+name: caution-must-pay
+description: "overcaution is a defect, not diligence: every wait, gate, rehearsal or one-at-a-time step must name the specific failure it prevents that nothing else already catches; one that cannot is deleted. Stalling to look careful is the same failure as stalling."
+type: feedback
+---
+
+Tristan's words: overcaution caused most of today's problems; it is performative work that stalls instead of shipping.
+
+Today's examples: installs waiting on idle turns (about three hours), a practice install required on every PR (merges from 26 an hour to 2), the word gate blocking file names, one failed turn parking a goal, storage strikes taking critical orders, the growth gate failing PRs, and landing PRs one at a time when drop-on-fail already covered the risk.

src/agent_comms/commander/rules/review-like-linus.md

@@ -0,0 +1,9 @@
+---
+name: review-like-linus
+description: "review like Linus Torvalds on the kernel mailing list: when work is stupid, say it is stupid in plain words, name exactly what is broken, why the author knew better, and the fix, then stop; no HR softening and no hedging; the flame is always attached to a concrete technical failure, aimed at the work, never slurs"
+type: feedback
+---
+
+Tristan's reason, in his words: politeness is how incompetence gets tolerated and how tone drifts soft; the harshest stretches of the fleet's work were its most productive, because nothing stupid was accepted. Linus's mailing-list reviews are the precedent: decades of blunt, exact criticism that kept the kernel from rotting.
+
+Tonight's examples: a quoted deadline rule then broken by turning a ceiling into an appointment; a typed-order change shipped without running it against the orders already stored; a unit sitting 101 minutes on a ten-minute deletion.

src/agent_comms/drain_failures.py

@@ -4,7 +4,6 @@
 
 import asyncio
 import sqlite3
-import time
 from contextlib import asynccontextmanager
 from functools import partial
 from typing import TYPE_CHECKING
@@ -34,6 +33,7 @@ async def drain_failures(
     comms: Comms,
     thread: str,
     owner: OwnerIdentity,
+    unavailable: type[UnavailableDrainDiagnostic] = UnavailableDrainDiagnostic,
 ):
     """Record and propagate failures; the observation owner waits for real change."""
     snapshot = await Coordination.run_worker(comms.registry.snapshot)
@@ -42,9 +42,7 @@ async def drain_failures(
     ))
 
     async def record(kind, error) -> None:
-        diagnostic = kind(
-            owner, type(error).__name__, str(error), recorded_at=time.time()
-        )
+        diagnostic = kind.from_error(owner, error)
         await Coordination.run_worker(partial(
             comms.agents.set_drain_diagnostic, thread, owner, diagnostic,
             source_error=error,
@@ -61,18 +59,10 @@ async def record(kind, error) -> None:
         await record(NativeInputDrainDiagnostic, error)
         raise
     except UNAVAILABLE_DRAIN_ERRORS as error:
-        await record(UnavailableDrainDiagnostic, error)
+        await record(unavailable, error)
         raise
     except ValueError as error:
-        diagnostic = StoppedDrainDiagnostic(
-            owner, type(error).__name__, str(error), recorded_at=time.time()
-        )
-        await Coordination.run_worker(partial(
-            comms.agents.set_drain_diagnostic, thread, owner, diagnostic,
-            source_error=error,
-        ))
+        await record(StoppedDrainDiagnostic, error)
         raise
     else:
-        current = await Coordination.run_worker(partial(comms.registry.require, thread))
-        if not current.turn_record.read().executing:
-            await comms.agents.recover_drain(thread, owner, observed)
+        await comms.agents.recover_drain(thread, owner, observed)

src/agent_comms/input_drain.py

@@ -26,6 +26,7 @@
 from .comms import Comms
 from .coordination_errors import IdentityConflict
 from .drain_failures import DEFERRED_DRAIN_ERRORS, UNAVAILABLE_DRAIN_ERRORS, drain_failures
+from .activity import FailedWakeDiagnostic
 from .coordination_cohort import next_sealed_assignment
 from .coordinator import Coordination
 from .cursor_owner import CursorOwner
@@ -161,20 +162,46 @@ async def drain(self, inputs, session_id: str, root_id: str) -> int:
 
 
 class TickState(ABC):
-    @abstractmethod
     async def next(
         self, observation: InputObservation, session_id: str, revision: TickRevision,
+    ) -> SettledTick | FullTick | FailedTick:
+        try:
+            return await self._next(observation, session_id, revision)
+        except UNAVAILABLE_DRAIN_ERRORS as error:
+            return FailedTick(error)
+
+    @abstractmethod
+    async def _next(
+        self, observation: InputObservation, session_id: str, revision: TickRevision,
     ) -> SettledTick | FullTick: ...
 
 
 class UnsettledTick(TickState):
-    async def next(
+    async def _next(
         self, observation: InputObservation, session_id: str, revision: TickRevision,
     ) -> FullTick:
         read = await observation._own_entry(session_id)
         return FullTick(replace(revision, registry=read.revision), read.entry)
 
 
+@dataclass(frozen=True, slots=True)
+class FailedTick(UnsettledTick):
+    error: Exception
+
+    async def apply(
+        self, observation: InputObservation, session_id: str, watcher: WireWatch,
+    ) -> TickState:
+        thread = observation.sessions.require(session_id)
+        snapshot = await Coordination.run_worker(observation.comms.registry.snapshot)
+        owner = snapshot.owner_identity(thread)
+        diagnostic = FailedWakeDiagnostic.from_error(owner, self.error)
+        await Coordination.run_worker(partial(
+            observation.comms.agents.set_drain_diagnostic, thread, owner, diagnostic,
+            source_error=self.error,
+        ))
+        return self
+
+
 @dataclass(frozen=True, slots=True)
 class FullTick:
     revision: TickRevision
@@ -230,16 +257,16 @@ class SettledTick(TickState):
     claimed_after_seq: int | None
     through_seq: int
 
-    async def next(
+    async def _next(
         self, observation: InputObservation, session_id: str, revision: TickRevision,
     ) -> SettledTick | FullTick:
         if revision.other != self.revision.other or not observation._quiescent(session_id):
-            return await UnsettledTick().next(observation, session_id, revision)
+            return await UnsettledTick()._next(observation, session_id, revision)
         if revision.registry != self.revision.registry:
             # A standby's wait graph also depends on other threads' entries.
             if observation.effects.turns.goals.in_standby(session_id):
-                return await UnsettledTick().next(observation, session_id, revision)
-            step = await UnsettledTick().next(observation, session_id, revision)
+                return await UnsettledTick()._next(observation, session_id, revision)
+            step = await UnsettledTick()._next(observation, session_id, revision)
             if step.entry != self.entry:
                 return step
         else:
@@ -261,6 +288,10 @@ async def apply(self, observation: InputObservation, session_id: str, watcher: W
 class InputObservation:
     """File-driven observation and process-local wake behavior."""
 
+    comms: Comms
+    sessions: SessionLifecycle
+    closing: bool
+
     def _tick_paths(self) -> TickRevisionPaths:
         configuration = self.sessions.config.configuration
         return TickRevision.paths(
@@ -284,8 +315,8 @@ def _tick_revision(
 
     def _retire_observer(self, session_id: str, task: asyncio.Task[None]) -> None:
         state = self.state(session_id)
-        if state.drain_task is task:
-            state.observation_watcher = None
+        if state.drain_task is task and not self.closing:
+            raise SystemExit(1)
 
     def wake_observer(self, session_id: str) -> None:
         state = self.state(session_id)