Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion src/agent_comms/history_views.py
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down
35 changes: 25 additions & 10 deletions src/agent_comms/presentation.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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
Expand All @@ -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:
Expand Down Expand Up @@ -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,),
Expand All @@ -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)})

Expand Down
4 changes: 4 additions & 0 deletions src/agent_comms/revision_views.py
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down
5 changes: 5 additions & 0 deletions src/agent_comms/todos.py
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand Down
217 changes: 217 additions & 0 deletions src/agent_comms/ui_model/commander.py
Original file line number Diff line number Diff line change
@@ -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)
Loading