diff --git a/src/agent_comms/history_views.py b/src/agent_comms/history_views.py index ebf03ca8e5..866f3bd320 100644 --- a/src/agent_comms/history_views.py +++ b/src/agent_comms/history_views.py @@ -196,7 +196,7 @@ def __init__( self.agents = agents self.transcripts = transcripts self.messaging = messaging - self.goals = goals + self.goals: Goals = goals self.ledger = ledger self._wire_lock_path = root / "wire" self.presentation = BusPresentation(bus, registry, channels.catalog, agents) diff --git a/src/agent_comms/presentation.py b/src/agent_comms/presentation.py index cf7adb2130..045e4e049b 100644 --- a/src/agent_comms/presentation.py +++ b/src/agent_comms/presentation.py @@ -20,10 +20,10 @@ from .channels import Channel, ChannelMemberIndex, Channels from .cli_result import CliResult from .coordination_errors import DMPaintedBoundChanged, StaleRevision +from .declared_family import DeclaredFamily from .display_order import ChannelSort, NameChannelSort from .display_references import DisplayReferences from .external_transcript import ExternalTranscript -from .declared_family import DeclaredFamily from .field_codec import FieldCodec, JsonShapeFamily, JsonShapeMember, projected from .goal_presentation import ThreadGoalExecution from .goal_waits import GoalWaits @@ -39,15 +39,27 @@ from .notification_progress import DeliveryProgress, HumanReadProgress, ReadDelivery from .pi_counts import PiCount from .private_path import FileRevision -from .ui_model.unread import UnreadPresentation, UnobservedUnread -from .read_basis import (AnonymousViewer, CapturedSelection, ChannelDisplayScope, DisplayViewer, - NamedViewer) +from .read_basis import ( + AnonymousViewer, + CapturedSelection, + ChannelDisplayScope, + DisplayViewer, + NamedViewer, +) from .read_ledger import NoReadNotice, ReadLedger, ReadNotice from .recorded_bus import RecordedBusIdentity from .registry_document import DEFAULT_ROSTER, RosterVisibility -from .revision_views import (STATUS_SOURCE, BusSourceView, ExternalTranscriptsView, - NotificationsView, RegistrationsView, RevisionView, - SidebarView, StoresView) +from .revision_views import ( + STATUS_SOURCE, + BusSourceView, + CommanderView, + ExternalTranscriptsView, + NotificationsView, + RegistrationsView, + RevisionView, + SidebarView, + StoresView, +) from .runtime_info import RuntimeInfo from .store_files import SharedLock, _store_lock from .thread_identity import ThreadIncarnation @@ -59,6 +71,7 @@ from .thread_status import ThreadStatus, ThreadStatusName from .threads import Thread from .tool_result import ToolResult +from .ui_model.unread import UnobservedUnread, UnreadPresentation from .wire_log import WireLog, WireSnapshot if TYPE_CHECKING: @@ -771,7 +784,9 @@ class WireRevision(ReadState): """One observation of every Comms store a view derives from, by name.""" registry: tuple[FileRevision, ...] = revision_source(lambda views: (views.registry.store.path,), - *STATUS_SOURCE, RegistrationsView, NotificationsView) + *STATUS_SOURCE, RegistrationsView, NotificationsView, CommanderView) + todos: tuple[FileRevision, ...] = revision_source(lambda views: views.goals.todos.source_paths(), + StoresView, CommanderView) catalog: tuple[FileRevision, ...] = revision_source(lambda views: (views.channels.catalog.path,), *STATUS_SOURCE, RegistrationsView) bus_log: tuple[FileRevision, ...] = revision_source(lambda views: (views.bus.log.path,), @@ -795,10 +810,10 @@ class WireRevision(ReadState): notifications: tuple[FileRevision, ...] = revision_source(lambda views: NotificationAssignment.handling_paths(views.root), *STATUS_SOURCE, NotificationsView) external_transcripts: tuple[FileRevision, ...] = revision_source(lambda views: ExternalTranscript.source_paths(views.root), - StoresView, ExternalTranscriptsView) + StoresView, ExternalTranscriptsView, CommanderView) # Notifications change once per expiry period, not per tick. expiry_tick: int = field(metadata={ - "views": (NotificationsView,), + "views": (NotificationsView, CommanderView), "differs": lambda before, now: ( before // WireRevision.expiry_seconds != now // WireRevision.expiry_seconds)}) diff --git a/src/agent_comms/revision_views.py b/src/agent_comms/revision_views.py index 29e6df6d9b..8e4e6cdf00 100644 --- a/src/agent_comms/revision_views.py +++ b/src/agent_comms/revision_views.py @@ -15,6 +15,10 @@ def differs(self, before: tuple[FileRevision, ...], current: tuple[FileRevision, return before != current +class CommanderView(RevisionView): + """Orders, publication clocks and expanded unit history.""" + + class RegistrationsView(RevisionView): pass diff --git a/src/agent_comms/todos.py b/src/agent_comms/todos.py index 3e155a81a1..78a06c4e0c 100644 --- a/src/agent_comms/todos.py +++ b/src/agent_comms/todos.py @@ -2916,6 +2916,11 @@ class TodoStore(TypedStore): filename: ClassVar[str] = "todos.sqlite3" rows: ClassVar[tuple[type[TypedTable], ...]] = (Todo, HuntRecord) + def source_paths(self) -> tuple[Path, ...]: + """SQLite's database and WAL both feed the observed order board.""" + path = self.path + return path, path.with_name(path.name + "-wal") + @contextmanager def writing(self) -> Iterator[sqlite3.Connection]: try: diff --git a/src/agent_comms/ui_model/commander.py b/src/agent_comms/ui_model/commander.py new file mode 100644 index 0000000000..bde35ec857 --- /dev/null +++ b/src/agent_comms/ui_model/commander.py @@ -0,0 +1,217 @@ +"""Commander cards prepared by Core from its existing order and progress owners.""" + +from __future__ import annotations + +import time +from collections.abc import Iterator +from dataclasses import dataclass +from pathlib import Path +from typing import TYPE_CHECKING + +from agent_comms.presentation import NoRevision, ReadState, WireRevision +from agent_comms.recorded_time import RecordedEventTime +from agent_comms.revision_views import CommanderView +from agent_comms.thread_identity import ThreadIncarnation +from agent_comms.todos import ( + MeasuredProgressSource, + OrderProgress, + ProgressPoint, + PullRequests, + PushedProgress, + Todo, +) +from agent_comms.turn_digests import LastTurns, ThreadTurns +from agent_comms.ui_model.observation import ( + InterestRequest, + Observation, + ObservationRequest, + ObservationStep, + Observed, + ReadRequest, + StoredRead, + UnreadableHistory, +) + +if TYPE_CHECKING: + from agent_comms.comms import Comms + from agent_comms.ui_model.observation import ObservationService + + +@dataclass(frozen=True) +class CommanderOrder: + todo: Todo + progress: OrderProgress + overdue: bool + published: tuple[ProgressPoint, ...] + pushes: tuple[PushedProgress, ...] + + @property + def line(self) -> str: + return ( + f"{'OVERDUE ' if self.overdue else ''}{self.todo.id}: " + f"{' '.join(self.todo.text.split())} | due {self.todo.deadline.clock()} | " + f"{self.todo.order.line}; {self.progress.line}" + ) + + @property + def cadence(self) -> str: + """Display actual publication clocks, never local-commit clocks or a stored forecast.""" + times = sorted({push.at for push in self.pushes}) + if len(times) < 2: + return f"{self.todo.id}: cadence unknown (need two recorded pushes)" + return ( + f"{self.todo.id}: {(times[-1] - times[-2]) / 60:.1f}m/push; " + f"last push {RecordedEventTime(times[-1]).clock()}" + ) + + +class CardDetail: + @property + def text(self) -> str: + return "Expand to read this unit's orders and recent turns." + + @property + def turn_lines(self) -> tuple[str, ...]: + return () + + +@dataclass(frozen=True) +class OpenCardDetail(CardDetail): + orders: tuple[CommanderOrder, ...] + turns: ThreadTurns + + @property + def text(self) -> str: + lines = [ + order.todo.order.checked().instructions(order.todo) + + "\nLatest work: " + + order.progress.line + for order in self.orders + ] + lines.append( + "Published commits:\n" + + "\n".join( + f"{order.todo.id}: {point.sha}" + for order in self.orders + for point in order.published + ) + ) + lines.append("Recent turns (newest first):") + return "\n\n".join(lines) + + @property + def turn_lines(self) -> tuple[str, ...]: + return tuple( + f"Turn {turn.turn_id}: {len(turn.events)} recorded events" for turn in self.turns.turns + ) or ("No recorded turns.",) + + +@dataclass(frozen=True) +class UnitCard: + unit: ThreadIncarnation + orders: tuple[CommanderOrder, ...] + detail: CardDetail + + @property + def line(self) -> str: + return self.unit.name + " | " + " ; ".join(order.line for order in self.orders) + + @property + def priority(self) -> tuple[bool, float, str]: + return ( + not any(order.overdue for order in self.orders), + min(order.todo.deadline.recorded_at("An order deadline") for order in self.orders), + self.unit.name.casefold(), + ) + + +@dataclass(frozen=True) +class CommanderModel: + commander: ThreadIncarnation + units: tuple[UnitCard, ...] + alarms: tuple[str, ...] + + @classmethod + def read(cls, comms: Comms, request: ObserveCommander) -> CommanderModel: + if Path(request.root).resolve() != comms.root.resolve(): + raise ValueError("The commander page belongs to a different Comms root; reopen it.") + commander = comms.registry.require(request.commander.name).incarnation + if commander != request.commander: + raise ValueError("The commander incarnation changed; reopen the page.") + todos = comms.goals.todos.held_for_commander(commander) + pulls, storage, now = PullRequests(todos), comms.registry.storage_rule, time.time() + grouped: dict[ThreadIncarnation, list[CommanderOrder]] = {} + for todo in todos: + source = MeasuredProgressSource.of(todo, storage, pulls) + grouped.setdefault(todo.held().owner_identity, []).append( + CommanderOrder( + todo, + OrderProgress.from_source(source), + todo.overdue(now), + tuple(source.published()), + source.pushes, + ) + ) + cards = [] + for unit, orders in grouped.items(): + if comms.registry.require(unit.name).incarnation != unit: + raise ValueError( + f"Order holder {unit.name} changed incarnation; refresh its assignment." + ) + current = tuple(orders) + detail: CardDetail = CardDetail() + if unit in request.expanded: + detail = OpenCardDetail(current, comms.transcripts.turns(unit.name, LastTurns(3))) + cards.append(UnitCard(unit, current, detail)) + alarms = tuple( + f"STALLED {order.todo.held().owner} {order.todo.id}: {order.progress.line}" + for card in cards + for order in card.orders + if order.todo.pending + and order.progress.stalled(order.todo.order.checked().at.recorded_at("An order"), now) + ) + return cls(commander, tuple(sorted(cards, key=lambda card: card.priority)), alarms) + + +@dataclass(frozen=True) +class ObserveCommander(InterestRequest): + root: str + commander: ThreadIncarnation + expanded: frozenset[ThreadIncarnation] = frozenset() + + def observation(self) -> CommanderReading: + return CommanderReading(self) + + +@dataclass(frozen=True) +class ReadCommanderIdentity(ReadRequest): + root: str + name: str + + def read(self, comms: Comms) -> ObserveCommander: + return ObserveCommander(self.root, comms.registry.require(self.name).incarnation) + + +class CommanderReading(Observation[ObserveCommander]): + def __init__(self, request: ObserveCommander): + super().__init__(request) + self.revision: ReadState = NoRevision() + + def read(self, comms: Comms, revision: WireRevision) -> Iterator[ObservationStep]: + if revision.changed_since(CommanderView(), self.revision): + model = UnreadableHistory.outcome(lambda: CommanderModel.read(comms, self.request)) + if model.answer(lambda value: True, lambda: False, lambda error: True): + self.revision = revision + yield CommanderObserved(revision, self.request, model) + + +@dataclass(frozen=True) +class CommanderObserved(Observed): + request: ObserveCommander + model: StoredRead[CommanderModel] + + +@dataclass(frozen=True) +class StopCommander(ObservationRequest): + def receive(self, service: ObservationService) -> None: + service.interest.remove(CommanderReading)