diff --git a/docs/agents/commander-page.md b/docs/agents/commander-page.md new file mode 100644 index 0000000000..b12ac04114 --- /dev/null +++ b/docs/agents/commander-page.md @@ -0,0 +1,110 @@ +# Commander board and derived commit cadence + +Issue #1495. Core #1937, paired Toad #1080. Owner: pi-cn-phase6. +These are source-paired changes, not an installation claim. + +## Working slice + +`/commander ` opens the ordinary Toad page. It shows held +orders, deadlines and delivery state, overdue first. Expanding one unit reads +its actual checkout commit and three original recent turns. Recorded failures, +stalled orders and retained refused sends appear in the alarms pane. + +The collapsed card also shows commit cadence, the forecast next-commit time, +its margin against the order deadline, and whether that forecast has already +been missed. Missing checkouts and insufficient/non-increasing samples remain +explicitly unknown. Delivery removes the forecast. + +**Interpretation:** commit cadence predicts the next commit, not completion. +There is no known amount of remaining work from which to calculate a completion +ETA. The page explicitly says `completion unknown`; no prediction is persisted. +Full row-27 prediction/outcome comparison remains unverified. pi-plan-claims +owns that row's declaration in #1992; no second parity registry was added here. +Strikes, review queue, story and sidebar decorations remain outside this slice. + +## Owners and deletions + +- `TodoStore.held_for_commander`, nominal holder identities and `Todo.overdue` + supply the canonical order selection and deadline meaning. +- `ProblemOrder.instructions` supplies order text. `CheckedProblemOrder.ordered_at` + owns recorded-time normalization; all numeric order-time consumers use it. +- `ProgressSource.progress` supplies one progress-selection algorithm. + `StatusProgressSource` inherits it and disables acceptance-check execution. +- Main's `ProgressCheckout` family owns available versus removed checkouts. + Its concrete classes supply commit details and bounded cadence evidence. + Both distance and cadence reads share `PresentCheckout.branch_commits` and + the `ProgressPoint` decoder. #1935's competing source changes were reverted. +- `UnitCard.momentum` derives `CommitMomentum` values from canonical orders and + observed commits. Unknown and forecast states are distinct types, not zero + timestamps standing in for missing observations. No new store or timer exists. +- `DeliveryProgress.failed_notifications` owns failure selection. `FailedProgress` + returns the original notification. The alarm reader's `isinstance` filter was + deleted; Toad has no failure-kind switch or notification copy. +- Actual GitHub command failures and timeouts become explicit unread progress. + Other Git, decode and source errors propagate; there is no ValueError catch-all. +- Existing observation interests, store/WAL revisions and expiry notifications + drive updates. Closing the page retires its interest. +- Toad renders the Core model through its existing Collapsible widgets. GoalBar + presence is derived from actual composition, not an optional cached child; + opening/clicking a bus conversation no longer raises `goal_bar` AttributeError. + +Used the existing refactor-audit Package parser across both production roots. +The declaration/consumer evidence reports zero parse omissions. Dynamic family +and MRO resolution was read semantically, not claimed as an AST behavioral proof. +Evidence is under `/home/ts/.cache/agent-scratch/pi-cn-phase6/`. + +## Physical application proof + +`physical-failure-owner/` contains the latest plain-st/Xvfb run, using normal +Toad, its real observation process, copied registered saved sessions, real +scratch-bus orders created through Core tools, and an actual retained refused +send. No provider was started, and no original session or uncertain input was +changed. `sources.txt` records provenance. + +The journey opened a real bus conversation, submitted `/commander claude-code`, +clicked one card, observed a tool-recorded delivery update without reopening, +and returned to the same conversation. Captures were inspected; no traceback +pane or unavailable-page error is present. + +- `level-0.txt/png`: collapsed cards, known and unknown cadence, alarms. +- `level-1.txt/png`: one expanded card and original turns; second card collapsed. +- `live-delivery.txt/png`: delivery updates the same open card and clears its stall. +- `result.txt`: successful continuous-journey result. + +Actual collapsed screen excerpt: + +```text +Commander claude-code | 2 units | Overdue first +P1: 3.0m/commit; next commit ~17:17 (132.1m before due); forecast missed by 133.2m; due 1.2m ago; completion unknown +P2: cadence unknown (need two increasing commit times); due in 29.8m; completion unknown +STALLED pi-cn-phase6 P1: pushed d07ee61b8 at 17:14, no PR +REFUSED SEND #1 claude-code -> unregistered-page-target: Target 'unregistered-page-target' is not a registered thread. +``` + +## Checks and review limits + +The stable source checkpoint passed **142 tests in 72.67s**. After removing the +failure-family filter, all **10 commander-model checks passed in 1.88s**, including +selection of the original failed notification. Tests ran in the background under +`timeout 300 ... -X faulthandler`, with default parallelism/coverage disabled. +The environment warns about the unavailable pytest-timeout plugin; the outer +timeout was enforced. Ruff for the new model and both diff checks pass. + +The retained full ratchet reports at the stable checkpoint both exit **1**, solely +for `SourceLines` (+465 Core, +205 tools), with no introduced findings. Rechecking +`ca9c4f881` after deleting the failure-family filter gives **src 1/tools 1**, solely ++474/+205 SourceLines, again with introduced findings empty. No budget, +suppression or checker change was made. Under the commander's ruling in message +3617, SourceLines-only failures do not block review. Reports name their exact +base/head; they are not a claim about a fresh merger integration head. + +The checking interpreter is the shared QA Python, with the paired Toad source and +existing stack dependencies visible. Toad now publishes its inline declarations +through `py.typed`; no copied Toad schema or stub was introduced. The installed +stack interpreter lacks the checking modules and was not modified. + +The required `order pr P243` create command was attempted once. It failed on its +derived missing `pi-cn-phase6--P243` directory; this task's assigned checkout is +`pi-cn-phase6--P1`, and #1937 already exists. The existing PR is updated in place +with metadata generated by its canonical order owner. GitHub currently reports +a conflict with current main; no merge-main or unrequested rebase was performed. diff --git a/src/agent_comms/commander/order.py b/src/agent_comms/commander/order.py index de6a972d2e..da4b514fdb 100644 --- a/src/agent_comms/commander/order.py +++ b/src/agent_comms/commander/order.py @@ -519,7 +519,7 @@ def first_calls(held, activity): unit, order = todo.assignment.owner, todo.order if not todo.pending: continue - acted = activity[unit].since(order.at.recorded_at("An order")) + acted = activity[unit].since(order.ordered_at) if not acted and time.time() < order.first_call_deadline.at: continue event = ObservedFirstCallEvent if acted else NotifiedFirstCallEvent @@ -590,7 +590,7 @@ def deadline_lines(pending: list[Todo], read: dict[str, OrderProgress], since: f now = time.time() late, stalled = [], [] for t in pending: - ordered, work = t.order.at.recorded_at('An order'), read[t.id] + ordered, work = t.order.ordered_at, read[t.id] if work.stalled_between(ordered, since, now): stalled.append(f'STALL {t.id} {t.holder}: {work.line} -> push it now or reassign') if not t.overdue(now): diff --git a/src/agent_comms/external_transcript.py b/src/agent_comms/external_transcript.py index d2eb3a25b0..f86b0b3af8 100644 --- a/src/agent_comms/external_transcript.py +++ b/src/agent_comms/external_transcript.py @@ -211,7 +211,7 @@ def unjudged_query(cls, thread, todo): tools = tuple(tool.declared_name for tool in ToolDeclaration.members_with(MutatingPathTool)) judged = tuple(verdict.tool_call_id for verdict in todo.verdicts) placeholders = lambda values: ",".join("?" for _ in values) - since = todo.order.at.recorded_at("An order") + since = todo.order.ordered_at # SQL NULL is the original undated fact, decoded through FieldCodec # as UnrecordedEventTime when a row is read. No time is substituted. table = cls.declared_name @@ -382,7 +382,7 @@ def unjudged(self, todo): for encoded, in rows: event = FieldCodec.decode(TranscriptEvent, FieldCodec.parse(encoded)) yield from event.edit_calls(todo.held().owner, todo.id, - todo.order.at.recorded_at("An order")) + todo.order.ordered_at) def unjudged_count(self, todo) -> int: where, parameters = ExternalEventRow.unjudged_query(self.transcript.lookup, todo) diff --git a/src/agent_comms/history_views.py b/src/agent_comms/history_views.py index a4c261c79b..c284b76548 100644 --- a/src/agent_comms/history_views.py +++ b/src/agent_comms/history_views.py @@ -174,6 +174,8 @@ def require(self, name: str) -> ThreadPresentation: class HistoryViews: + goals: Goals + def __init__( self, root: Path, diff --git a/src/agent_comms/notification_progress.py b/src/agent_comms/notification_progress.py index 52f7d9a9e4..36087c4922 100644 --- a/src/agent_comms/notification_progress.py +++ b/src/agent_comms/notification_progress.py @@ -22,6 +22,11 @@ class DeliveryProgress(ThreadPresentationStatus): def notification_label(self, notification: MessageNotification) -> str: return self.label + def failed_notifications( + self, notification: MessageNotification + ) -> tuple[MessageNotification, ...]: + return () + class PendingDelivery: delivered: ClassVar[bool] = False @@ -87,6 +92,11 @@ class InterruptedProgress(ContextualProgress): class FailedProgress(ContextualProgress): label = "Failed" + def failed_notifications( + self, notification: MessageNotification + ) -> tuple[MessageNotification, ...]: + return (notification,) + class EngagedProgress(ContextualProgress, WorkingThreadPresentationStatus): priority = 1 diff --git a/src/agent_comms/presentation.py b/src/agent_comms/presentation.py index d7784501c8..c8b49a7eb5 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,), @@ -793,12 +808,12 @@ class WireRevision(ReadState): relationships: tuple[FileRevision, ...] = revision_source(lambda views: (relationships.RelationshipStore.path_in(views.root),), *STATUS_SOURCE) notifications: tuple[FileRevision, ...] = revision_source(lambda views: NotificationAssignment.handling_paths(views.root), - *STATUS_SOURCE, NotificationsView) + *STATUS_SOURCE, NotificationsView, CommanderView) 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..c5cb141606 100644 --- a/src/agent_comms/revision_views.py +++ b/src/agent_comms/revision_views.py @@ -36,6 +36,10 @@ class SidebarView(RevisionView): """Native unread progress updates counts without rebuilding thread presentations.""" +class CommanderView(RevisionView): + """Orders, retained deliveries, recent turns and time-derived commander attention.""" + + class ExternalTranscriptsView(RevisionView): """CLI drafts and finished items, independently of coordination status.""" diff --git a/src/agent_comms/todos.py b/src/agent_comms/todos.py index 6a9d18f50a..ece47c990d 100644 --- a/src/agent_comms/todos.py +++ b/src/agent_comms/todos.py @@ -20,7 +20,7 @@ from functools import cache from importlib.resources import files from pathlib import Path -from typing import TYPE_CHECKING, ClassVar, Final, Self +from typing import TYPE_CHECKING, ClassVar, Self from uuid import uuid4 from . import skills @@ -244,8 +244,7 @@ def read(self, repo: str, state: str, order: ProblemOrder) -> tuple[PullRequest, "--jq", '.data | map_values(.nodes | map([.number, .headRefName, .headRefOid,' ' .mergeCommit.oid // "", .mergedAt // "", .updatedAt, .body]))'), capture_output=True, text=True, timeout=60) - if listed.returncode: - raise ValueError(listed.stderr.strip() or f"GitHub order lookup in {repo} failed.") + listed.check_returncode() rows = FieldCodec.decode(dict[str, tuple[tuple[int, str, str, str, str, str, str], ...]], FieldCodec.parse(listed.stdout)) for alias, candidate in aliases.items(): @@ -258,6 +257,83 @@ def _clock(at: float) -> str: return time.strftime("%H:%M", time.localtime(at)) +@dataclass(frozen=True) +class ProgressPoint: + """A revision and its time in the order's progress; Git rows use committer time.""" + sha: str + at: float + format: ClassVar[str] = "%H %ct" + + @classmethod + def listed(cls, line: str) -> ProgressPoint: + sha, at = line.split() + return cls(sha, float(at)) + + +@dataclass(frozen=True) +class CommitMomentum: + """A next-commit forecast from observed commits, not a completion estimate. + + The card derives this value on read; nothing is persisted. Without a known + amount of remaining work, commit cadence cannot predict order completion. + """ + todo: Todo + observed_at: float + window: ClassVar[int] = 2 + + @classmethod + def from_commits(cls, todo: Todo, commits: tuple[ProgressPoint, ...], + observed_at: float) -> CommitMomentum: + if len(commits) < cls.window or commits[-1].at <= commits[0].at: + return UnknownCommitMomentum(todo, observed_at) + return ForecastCommitMomentum(todo, observed_at, commits[0], commits[-1]) + + @property + @abstractmethod + def forecast(self) -> str: ... + + @property + def line(self) -> str: + if not self.todo.pending: + return f"{self.todo.id}: momentum not applicable; order delivered" + due = self.todo.deadline.recorded_at("An order deadline") - self.observed_at + deadline = f"due in {due / 60:.1f}m" if due >= 0 else f"due {-due / 60:.1f}m ago" + return f"{self.todo.id}: {self.forecast}; {deadline}; completion unknown" + + +class UnknownCommitMomentum(CommitMomentum): + @property + def forecast(self) -> str: + return "cadence unknown (need two increasing commit times)" + + +@dataclass(frozen=True) +class ForecastCommitMomentum(CommitMomentum): + first: ProgressPoint + last: ProgressPoint + + @property + def interval(self) -> float: + return self.last.at - self.first.at + + @property + def next_at(self) -> float: + return self.last.at + self.interval + + @property + def deadline_margin(self) -> float: + return self.todo.deadline.recorded_at("An order deadline") - self.next_at + + @property + def forecast(self) -> str: + margin = self.deadline_margin + relation = "before" if margin >= 0 else "after" + late = self.observed_at - self.next_at + missed = f"; forecast missed by {late / 60:.1f}m" if late > 0 else "" + return (f"{self.interval / 60:.1f}m/commit; next commit ~{_clock(self.next_at)} " + f"({abs(margin) / 60:.1f}m {relation} due){missed}") + + @dataclass(frozen=True) class DistanceReading: at: float @@ -353,6 +429,12 @@ def commits(self, since: float, *excluded: str) -> Iterator[tuple[str, float]]: @abstractmethod def readings(self, source: ProgressSource) -> DistanceReadings: ... + @abstractmethod + def cadence_commits(self, since: float) -> tuple[ProgressPoint, ...]: ... + + @abstractmethod + def latest_commit(self) -> str: ... + class MissingCheckout(ProgressCheckout): """A retired checkout contributes no local history, not a failed Git read.""" @@ -363,6 +445,12 @@ def commits(self, since: float, *excluded: str) -> Iterator[tuple[str, float]]: def readings(self, source: ProgressSource) -> DistanceReadings: return DistanceReadings() + def cadence_commits(self, since: float) -> tuple[ProgressPoint, ...]: + return () + + def latest_commit(self) -> str: + return "Checkout unavailable; no local commit evidence." + class PresentCheckout(ProgressCheckout): def git(self, *args: str) -> str: @@ -383,13 +471,26 @@ def commits(self, since: float, *excluded: str) -> Iterator[tuple[str, float]]: def readings(self, source: ProgressSource) -> DistanceReadings: """The order's base and newest branch commit bound its distance trend.""" base = self.git("merge-base", "HEAD", "origin/HEAD").strip() - after = time.strftime('%Y-%m-%dT%H:%M:%SZ', time.gmtime(source.since)) - commits = [(sha, float(at)) for sha, at in (line.split() for line in reversed(self.git( - "log", "--first-parent", f"--since={after}", "--format=%H %ct", "HEAD", - "--not", base).splitlines()))] - points = [(base, source.since), *commits][-2:] - return DistanceReadings( - tuple(source.reading(self, base, sha, at) for sha, at in points)) + commits = self.branch_commits(source.since, base, 2) + points = [ProgressPoint(base, source.since), *commits][-2:] + return DistanceReadings(tuple( + source.reading(self, base, point.sha, point.at) for point in points)) + + def latest_commit(self) -> str: + return self.git("log", "-1", "--format=%h %s").strip() + + def cadence_commits(self, since: float) -> tuple[ProgressPoint, ...]: + base = self.git("merge-base", "HEAD", "origin/HEAD").strip() + return self.branch_commits(since, base, CommitMomentum.window) + + def branch_commits(self, since: float, base: str, limit: int) -> tuple[ProgressPoint, ...]: + """Decode the bounded order-time branch window once for all readers.""" + after = time.strftime('%Y-%m-%dT%H:%M:%SZ', time.gmtime(since)) + return tuple(point for line in reversed(self.git( + "log", f"--max-count={limit}", "--first-parent", f"--since={after}", + f"--format={ProgressPoint.format}", "HEAD", "--not", base).splitlines()) + if (point := ProgressPoint.listed(line)).at >= since) + @dataclass(frozen=True) @@ -407,7 +508,19 @@ def of(cls, todo: Todo, storage: StorageRule, pulls: PullRequests) -> Self: @property def since(self) -> float: - return self.todo.order.at.recorded_at("An order") + return self.todo.order.ordered_at + + def progress(self) -> OrderProgress: + """One selection algorithm for measured tools and read-only status views.""" + members = sorted(OrderProgress.members_with(FoundProgress), + key=lambda member: member.precedence) + try: + return next((progress for member in members for progress in member.found(self)), + NoCommitProgress()) + except subprocess.CalledProcessError as unread: + return UnreadProgress(unread.stderr.strip() or str(unread)) + except subprocess.TimeoutExpired as unread: + return UnreadProgress(str(unread)) def requests(self, state: str) -> list[PullRequest]: """Explicit PR references survive checkout removal or branch reuse.""" @@ -417,9 +530,16 @@ def requests(self, state: str) -> list[PullRequest]: def commits(self, *excluded: str) -> Iterator[tuple[str, float]]: return self.checkout.commits(self.since, *excluded) + def latest_commit(self) -> str: + """The checkout's latest actual commit, even when it predates this order.""" + return self.checkout.latest_commit() + def readings(self) -> DistanceReadings: return self.checkout.readings(self) + def cadence_commits(self) -> tuple[ProgressPoint, ...]: + return self.checkout.cadence_commits(self.since) + def reading(self, checkout: PresentCheckout, base: str, sha: str, at: float) -> DistanceReading: """Keep the check command with its measured or missing distance.""" check = self.todo.order.check @@ -429,6 +549,14 @@ def reading(self, checkout: PresentCheckout, base: str, sha: str, at: float) -> self.todo.order.untouched(touched)), check.command) +@dataclass(frozen=True) +class StatusProgressSource(ProgressSource): + """A display reads git/PR status, never executes the order's acceptance check.""" + + def readings(self) -> DistanceReadings: + return DistanceReadings() + + @dataclass(frozen=True) class OrderProgress(DeclaredFamily, affix="Progress"): """Where an order's work stands, derived on each read from its unit's checkout and PRs. @@ -438,13 +566,7 @@ class OrderProgress(DeclaredFamily, affix="Progress"): @classmethod def read(cls, todo: Todo, storage: StorageRule, pulls: PullRequests) -> OrderProgress: """The first found member by its declared precedence; with none found, no commit yet.""" - source = ProgressSource.of(todo, storage, pulls) - members = sorted(cls.members_with(FoundProgress), key=lambda member: member.precedence) - try: - return next((progress for member in members for progress in member.found(source)), - NoCommitProgress()) - except subprocess.TimeoutExpired as unread: - return UnreadProgress(str(unread)) + return ProgressSource.of(todo, storage, pulls).progress() @property @abstractmethod @@ -1119,6 +1241,14 @@ def pull_request_ids(cls, body: str) -> tuple[str, ...]: return tuple(line.removeprefix(cls.pull_request_label).strip() for line in body.splitlines() if line.startswith(cls.pull_request_label)) + @property + def ordered_at(self) -> float: + raise ValueError("This problem has no checked order time.") + + def instructions(self, todo: Todo) -> str: + return "\n".join((f"Problem: {todo.id}: {todo.text}", + *self.finding.instructions(), self.line)) + def require_transfer(self, todo: Todo) -> None: raise ValueError("This problem cannot receive another order.") @@ -1157,10 +1287,6 @@ class UnorderedProblemOrder(ProblemOrder): def require_transfer(self, todo: Todo) -> None: pass - def instructions(self, todo: Todo) -> str: - return "\n".join((f"Problem: {todo.id}: {todo.text}", - *self.finding.instructions(), "No order")) - def linked(self, order: PendingProblemOrder) -> PendingProblemOrder: return replace(order, finding=self.finding) @@ -1196,6 +1322,10 @@ class CheckedProblemOrder(ProblemOrder, declared_name=None): default_factory=NoOrderDeletion, metadata={"wire_omit_default": True}) serves: type[OrderPhase] = field(default=UnplannedPhase, metadata={"wire_omit_default": True}) + @property + def ordered_at(self) -> float: + return self.at.recorded_at("An order") + def __post_init__(self): if not self.edits or not self.banned: raise ValueError("An order requires edits and banned actions.") @@ -1305,7 +1435,7 @@ def waiting_on_install(self, todo: Todo, read: dict[str, OrderProgress]) -> tupl class PendingProblemOrder(CheckedProblemOrder): def reissued(self, at: float) -> PendingProblemOrder: window = (self.first_call_deadline.recorded_at("A first-call deadline") - - self.at.recorded_at("An order")) + - self.ordered_at) return replace(self, id=uuid4().hex[:12], at=RecordedEventTime(at), first_call_deadline=RecordedEventTime(at + window), first_call=AwaitingFirstCall()) @@ -2055,7 +2185,7 @@ def live_strikes(self, unit: str | None = None) -> tuple[StrikeTodo, ...]: def struck_on_order(self, unit: str) -> bool: """A live strike on ``unit`` recorded since the current order was sent; a send-back that replaces this order is that same miss, so it counts no second strike.""" - sent = self.order.at.recorded_at("An order") + sent = self.order.ordered_at return any(recorded.at.recorded_at("A strike") >= sent for recorded in self.live_strikes(unit)) @@ -2195,6 +2325,11 @@ class TodoStore(TypedStore): filename: ClassVar[str] = "todos.sqlite3" rows: ClassVar[tuple[type[TypedTable], ...]] = (Todo, HuntRecord) + def source_paths(self) -> tuple[Path, ...]: + """Both committed SQLite locations can change an observed order.""" + path = self.path + return path, path.with_name(path.name + "-wal") + def record_hunt(self, commander: ThreadIncarnation, hunt: Hunt) -> HuntRecord: hunt.require_commander(self, commander) recorded = HuntRecord(uuid4().hex, commander, hunt) @@ -2216,7 +2351,7 @@ def resolve_order(self, commander: ThreadIncarnation, unit: str, message: str) - def drop(self, commander: ThreadIncarnation, unit: str, message: str, by: str) -> Todo: old = self.resolve_order(commander, unit, message) newer = self.resolve_order(commander, unit, by) - if not old.pending or not newer.pending or newer.order.at.at <= old.order.at.at: + if not old.pending or not newer.pending or newer.order.ordered_at <= old.order.ordered_at: raise ValueError('Only a live newer order to the same unit replaces an old one.') with self.writing() as db: current = self._current(db, old.id) diff --git a/src/agent_comms/tools.py b/src/agent_comms/tools.py index f07fe9a1eb..1c8526c646 100644 --- a/src/agent_comms/tools.py +++ b/src/agent_comms/tools.py @@ -755,7 +755,7 @@ def readers(cls, comms: Comms, commander: str, read_type=OrderEditsRead): raise ToolRefusedError( f"Order {todo.id} holder incarnation no longer exists" ) - yield read_type(comms.root, holder, TurnsSince(todo.order.at.recorded_at("An order")), todo) + yield read_type(comms.root, holder, TurnsSince(todo.order.ordered_at), todo) @classmethod def count(cls, comms: Comms, commander: str) -> int: @@ -955,7 +955,7 @@ class CommsOrderWorkTool(ReadTool, ToolRequest): def apply(self, comms: Comms, caller: ToolCaller) -> OrderWork: todo = comms.goals.todos.get(self.problem) - since = todo.order.at.recorded_at("An order") + since = todo.order.ordered_at turns = comms.transcripts.turns(todo.held().owner, TurnsSince(since)) return OrderWork.read(self.problem, turns.thread, since, turns.turns) diff --git a/src/agent_comms/transcript_events.py b/src/agent_comms/transcript_events.py index 2f342e73fe..912870af62 100644 --- a/src/agent_comms/transcript_events.py +++ b/src/agent_comms/transcript_events.py @@ -101,6 +101,10 @@ def imported(self) -> ImportedMessage | None: """This event as portable text in an imported conversation; activity is not conversation.""" return None + def turn_line(self) -> str: + """A compact account of the unit's work; prompts and silent evidence add none.""" + return "" + def record_call(self, calls: dict[str, OrderCall], since: float) -> None: """Only tool events change a unit's calls.""" @@ -256,6 +260,9 @@ def imported(self) -> ImportedMessage: class AssistantTranscript(AgentTextTranscript): + def turn_line(self) -> str: + return " ".join(self.text.split())[:240] + @property def category(self) -> type[MessageCategory]: return OutboundCategory if self.routed else AgentCategory @@ -351,6 +358,9 @@ def kind(self) -> str: @dataclass(frozen=True, kw_only=True) class ToolStartTranscript(ToolTranscript): + def turn_line(self) -> str: + return self.title + def edit_calls(self, thread: str, problem: str, since: float) -> tuple[tool_declaration.EditCall, ...]: if self.before(since) or self.tool is None: return () @@ -440,6 +450,11 @@ class TurnDigest: turn_id: str events: tuple[TranscriptEvent, ...] + @property + def line(self) -> str: + work = tuple(line for event in self.events if (line := event.turn_line())) + return f"{self.turn_id}: " + (" ยท ".join(work[-3:])[:400] or "No work recorded yet.") + def before(self, at: float) -> bool: """The turn's last dated event precedes ``at``; an undated turn is never before.""" dated = [event for event in self.events if event.dated] diff --git a/src/agent_comms/turn_digests.py b/src/agent_comms/turn_digests.py index 69ba2f14f7..ded5afa013 100644 --- a/src/agent_comms/turn_digests.py +++ b/src/agent_comms/turn_digests.py @@ -119,7 +119,7 @@ class OrderRead[Result](TranscriptRead[Result]): def edits_from_turns(self, turns: tuple[TurnDigest, ...]) -> tuple[EditCall, ...]: judged = {verdict.tool_call_id for verdict in self.todo.verdicts} - since = self.todo.order.at.recorded_at("An order") + since = self.todo.order.ordered_at edits = {edit.id: edit for turn in reversed(turns) for event in turn.events for edit in event.edit_calls(self.thread.name, self.todo.id, since) diff --git a/src/agent_comms/ui_model/commander.py b/src/agent_comms/ui_model/commander.py new file mode 100644 index 0000000000..0a3396046b --- /dev/null +++ b/src/agent_comms/ui_model/commander.py @@ -0,0 +1,292 @@ +"""Read-only commander cards, derived from Core's order and transcript owners. + +The observer prepares these models. A frontend supplies only commander identity +and expansion interest; it never reads stores, runs checks or decides order state. +""" + +from __future__ import annotations + +import time +from abc import abstractmethod +from dataclasses import dataclass +from pathlib import Path +from typing import TYPE_CHECKING + +from agent_comms.coordination_tables.refusals import SendRefusal +from agent_comms.declared_family import DeclaredFamily +from agent_comms.presentation import NoRevision, ReadState, WireRevision +from agent_comms.revision_views import CommanderView +from agent_comms.thread_identity import ThreadIncarnation +from agent_comms.todos import ( + CommitMomentum, + OrderProgress, + ProgressPoint, + PullRequests, + StatusProgressSource, + Todo, +) +from agent_comms.turn_digests import LastTurns, ThreadTurns +from agent_comms.ui_model.observation import ( + InterestRequest, + Observation, + ObservationRequest, + Observed, + ReadRequest, + StoredRead, + UnreadableHistory, +) + +if TYPE_CHECKING: + from collections.abc import Iterator + + from agent_comms.comms import Comms + from agent_comms.ui_model.observation import ObservationService, ObservationStep + + +@dataclass(frozen=True) +class CommanderOrder: + todo: Todo + progress: OrderProgress + overdue: bool + commits: tuple[ProgressPoint, ...] + observed_at: float + + @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.state}" + ) + + +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 + latest_commits: tuple[str, ...] + + @property + def text(self) -> str: + lines = [ + order.todo.order.instructions(order.todo) + "\nLatest work: " + order.progress.state + for order in self.orders + ] + lines.append("Latest checkout commits:\n" + "\n".join(self.latest_commits)) + lines.append("Recent turns (newest first):") + return "\n\n".join(lines) + + @property + def turn_lines(self) -> tuple[str, ...]: + return tuple(turn.line 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 momentum(self) -> tuple[CommitMomentum, ...]: + """Derived from order times and commit evidence; never a stored score.""" + return tuple( + CommitMomentum.from_commits(order.todo, order.commits, order.observed_at) + 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 CommanderAlarm: + line: str + + +class CommanderAlarmSource(DeclaredFamily, affix="AlarmSource"): + """New retained alarm sources extend the model, never the page's rendering. + + Retained pre-send refusals and failed deliveries have different source owners. + Each source supplies its own lines; the frontend has no kind switches. + """ + + @classmethod + @abstractmethod + def read( + cls, + comms: Comms, + commander: ThreadIncarnation, + orders: tuple[CommanderOrder, ...], + now: float, + ) -> tuple[CommanderAlarm, ...]: ... + + +class StalledOrderAlarmSource(CommanderAlarmSource): + @classmethod + def read( + cls, + comms: Comms, + commander: ThreadIncarnation, + orders: tuple[CommanderOrder, ...], + now: float, + ) -> tuple[CommanderAlarm, ...]: + return tuple( + CommanderAlarm( + f"STALLED {order.todo.held().owner} {order.todo.id}: {order.progress.state}" + ) + for order in orders + if order.todo.pending and order.progress.stalled(order.todo.order.ordered_at, now) + ) + + +class FailedDeliveryAlarmSource(CommanderAlarmSource): + @classmethod + def read( + cls, + comms: Comms, + commander: ThreadIncarnation, + orders: tuple[CommanderOrder, ...], + now: float, + ) -> tuple[CommanderAlarm, ...]: + presentation = comms.views.thread_presentations((commander.name,)).require(commander.name) + return tuple( + CommanderAlarm(f"RECORDED FAILED DELIVERY to {failed.recipient}: {failed.label}") + for notice in presentation.sent_notifications + for failed in notice.state.failed_notifications(notice) + ) + + +class RefusedSendAlarmSource(CommanderAlarmSource): + @classmethod + def read( + cls, + comms: Comms, + commander: ThreadIncarnation, + orders: tuple[CommanderOrder, ...], + now: float, + ) -> tuple[CommanderAlarm, ...]: + return tuple( + CommanderAlarm( + f"REFUSED SEND #{refusal.id} {refusal.sender} -> {refusal.target}: {refusal.reason}" + ) + for refusal in SendRefusal.recent(comms.root, sender=commander.name) + ) + + +@dataclass(frozen=True) +class CommanderModel: + commander: ThreadIncarnation + units: tuple[UnitCard, ...] + alarms: tuple[CommanderAlarm, ...] + + @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: + unit = todo.held().owner_identity + source = StatusProgressSource.of(todo, storage, pulls) + grouped.setdefault(unit, []).append( + CommanderOrder( + todo, source.progress(), todo.overdue(now), source.cadence_commits(), now + ) + ) + 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: + commits = [] + for order in current: + source = StatusProgressSource.of(order.todo, storage, pulls) + commit = source.latest_commit() + commits.append(f"{order.todo.id}: {commit}") + detail = OpenCardDetail( + current, comms.transcripts.turns(unit.name, LastTurns(3)), tuple(commits) + ) + cards.append(UnitCard(unit, current, detail)) + all_orders = tuple(order for card in cards for order in card.orders) + alarms = tuple( + alarm + for source in CommanderAlarmSource.members_with(CommanderAlarmSource) + for alarm in source.read(comms, commander, all_orders, 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): + """Resolve the external command spelling once, in the existing read process.""" + + 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)) + # Contention is retried on the next observer turn, not published as an empty board. + ready = model.answer(lambda value: True, lambda: False, lambda error: True) + if ready: + 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) diff --git a/src/agent_comms/ui_model/observation.py b/src/agent_comms/ui_model/observation.py index 3ea0fa1c48..a4ab2d4c35 100644 --- a/src/agent_comms/ui_model/observation.py +++ b/src/agent_comms/ui_model/observation.py @@ -42,7 +42,7 @@ CoordinationReadUnavailable, StaleRevision, ) -from agent_comms.errors import RelationViolationError +from agent_comms.errors import RelationViolationError, ToolRefusedError from agent_comms.field_codec import FieldCodec from agent_comms.presentation import NoRevision, ReadState, WireRevision from agent_comms.registry_document import DEFAULT_ROSTER, RosterVisibility @@ -346,7 +346,8 @@ def answer(self, comms: Comms) -> ReadAnswered | ReadRefused | ReadBusy: except self.deferred_reads: return ReadBusy(self.request_id) return ReadAnswered(self.request_id, value) - except (CoordinationReadUnavailable, RelationViolationError, UnreadableHistory, StaleRevision) as error: + except (CoordinationReadUnavailable, RelationViolationError, UnreadableHistory, + StaleRevision, ToolRefusedError) as error: return self.refused(error) diff --git a/tests/test_commander_model.py b/tests/test_commander_model.py new file mode 100644 index 0000000000..eed930994b --- /dev/null +++ b/tests/test_commander_model.py @@ -0,0 +1,117 @@ +"""Real Core stores back commander models; the page never executes an order check.""" +from dataclasses import replace + +import pytest + +from agent_comms.audience_manifest import FrozenRecipient +from agent_comms.notification_progress import FailedProgress, SentProgress +from agent_comms.presentation import MessageNotification +from agent_comms.recorded_time import RecordedEventTime +from agent_comms.todos import CommitMomentum, ForecastCommitMomentum, UnknownCommitMomentum, ProgressPoint, ProgressSource, PullRequests, StatusProgressSource +from agent_comms.ui_model.commander import ( + CommanderModel, + ObserveCommander, + OpenCardDetail, + ReadCommanderIdentity, + CommanderOrder, + CardDetail, + UnitCard, +) +from agent_comms.ui_model.observation import ReadAnswered, ReadRefused +from test_commander_problem_orders import create + + +def test_board_uses_order_state_and_requests_details_only_when_opened(commander): + comms, caller = commander + lead = comms.registry.require(caller.name).incarnation + unit = comms.registry.require('worker').incarnation + create(commander, 'P1', deadline=1000) + create(commander, 'P2', deadline=2000) + request = ObserveCommander(str(comms.root), lead) + model = CommanderModel.read(comms, request) + assert len(model.units) == 1 + card, = model.units + assert card.unit == unit + assert 'P1' in card.line and 'P2' in card.line and 'OVERDUE' in card.line + assert 'printf' not in card.line and 'Repair the existing owner' not in card.detail.text + assert len(model.alarms) == 2 + opened = CommanderModel.read(comms, replace(request, expanded=frozenset((unit,)))) + assert isinstance(opened.units[0].detail, OpenCardDetail) + assert 'Repair the existing owner' in opened.units[0].detail.text + assert 'Latest work:' in opened.units[0].detail.text + assert 'Recent turns' in opened.units[0].detail.text + + +def test_status_source_never_requests_acceptance_measurement(commander, monkeypatch): + comms, caller = commander + todo = create(commander, 'P1') + def forbidden(self): + pytest.fail('A read-only commander page executed an acceptance measurement') + monkeypatch.setattr(ProgressSource, 'readings', forbidden) + source = StatusProgressSource.of(todo, comms.registry.storage_rule, PullRequests((todo,))) + assert not source.readings().readings + model = CommanderModel.read(comms, ObserveCommander( + str(comms.root), comms.registry.require(caller.name).incarnation)) + assert model.units[0].orders[0].progress.state == 'no commit since the order' + + +def test_card_derives_commit_cadence_against_deadline_without_storing_momentum(commander): + comms, caller = commander + todo = replace(create(commander, 'P1'), deadline=RecordedEventTime(300)) + points = (ProgressPoint('first', 100), ProgressPoint('second', 160)) + order = CommanderOrder(todo, ProgressSource.of(todo, comms.registry.storage_rule, PullRequests((todo,))).progress(), + False, points, 170) + card = UnitCard(comms.registry.require('worker').incarnation, (order,), CardDetail()) + momentum, = card.momentum + assert 'momentum' not in card.__dataclass_fields__ + assert isinstance(momentum, ForecastCommitMomentum) + assert momentum.interval == 60 + assert momentum.next_at == 220 + assert momentum.deadline_margin == 80 + assert '1.3m before due' in momentum.line + assert 'completion unknown' in momentum.line + late = replace(momentum, todo=replace(todo, deadline=RecordedEventTime(200)), observed_at=250) + assert '0.3m after due' in late.line + assert 'forecast missed by 0.5m' in late.line + + +@pytest.mark.parametrize('points', [(), (ProgressPoint('one', 100),), + (ProgressPoint('one', 100), ProgressPoint('two', 100)), + (ProgressPoint('one', 160), ProgressPoint('two', 100))]) +def test_unknown_commit_cadence_does_not_invent_forecast(commander, points): + todo = create(commander, 'P1') + momentum = CommitMomentum.from_commits(todo, points, 170) + assert isinstance(momentum, UnknownCommitMomentum) + assert 'cadence unknown' in momentum.line + assert 'next commit ~' not in momentum.line + + +def test_delivery_progress_selects_its_original_failed_notification(): + sent = MessageNotification(FrozenRecipient('worker', 'worker'), 'lead', SentProgress(1)) + assert sent.state.failed_notifications(sent) == () + failed = replace(sent, state=FailedProgress(sent.state)) + selected, = failed.state.failed_notifications(failed) + assert selected is failed + assert selected.label == 'Failed' + + +def test_commander_identity_is_resolved_or_refused_in_the_read_process(commander): + comms, caller = commander + answer = ReadCommanderIdentity(str(comms.root), caller.name).answer(comms) + assert isinstance(answer, ReadAnswered) + assert answer.value == ObserveCommander( + str(comms.root), comms.registry.require(caller.name).incarnation) + refusal = ReadCommanderIdentity(str(comms.root), 'absent-commander').answer(comms) + assert isinstance(refusal, ReadRefused) + assert 'absent-commander' in str(refusal.error) + + +def test_commander_read_rejects_root_and_incarnation_changes(commander): + comms, caller = commander + lead = comms.registry.require(caller.name).incarnation + request = ObserveCommander(str(comms.root / 'elsewhere'), lead) + with pytest.raises(ValueError, match='different Comms root'): + CommanderModel.read(comms, request) + with pytest.raises(ValueError, match='incarnation changed'): + CommanderModel.read(comms, replace(request, root=str(comms.root), + commander=replace(lead, created_at=lead.created_at + 1))) diff --git a/tests/test_order_progress.py b/tests/test_order_progress.py index b676797076..53578d51f5 100644 --- a/tests/test_order_progress.py +++ b/tests/test_order_progress.py @@ -163,7 +163,7 @@ def test_a_merged_order_still_awaits_its_record(checkout): def test_unread_github_is_explicit(checkout): pulls = Listed() - pulls.states = ValueError('gh: not logged in') + pulls.states = subprocess.CalledProcessError(1, 'gh api graphql', stderr='gh: not logged in') progress = read(checkout, pulls) assert progress == UnreadProgress('gh: not logged in') diff --git a/tests/test_turn_digests.py b/tests/test_turn_digests.py index 105ca45654..d1c9a6f0be 100644 --- a/tests/test_turn_digests.py +++ b/tests/test_turn_digests.py @@ -39,7 +39,8 @@ def test_external_turns_preserve_interleaving_and_the_whole_selected_turn(tmp_pa source = comms.transcripts.external('cli') for turn, item, text in [('old', 'a', 'older'), ('one', 'b', 'first'), ('two', 'c', 'other'), ('one', 'd', 'continued')]: - source.finish(turn_id=turn, item_id=item, events=(AssistantTranscript(text),)) + source.finish(turn_id=turn, item_id=item, + events=(AssistantTranscript(text, position=item),)) answer = endpoint(comms, 'cli', 2) assert [turn['turn_id'] for turn in answer['turns']] == ['one', 'two'] assert [event['text'] for event in answer['turns'][0]['events']] == ['first', 'continued'] @@ -64,6 +65,8 @@ def test_native_turns_keep_original_calls_results_and_input_identity(tmp_path): dict(type='message', id='a2', parentId='result', message=dict( role='assistant', content=[dict(type='text', text='done')], stopReason='stop')), ] + for row in rows: + row['timestamp'] = '2026-10-16T12:00:00Z' source.write_text(''.join(json.dumps(row) + '\n' for row in rows)) source.chmod(0o600) comms.registry.declare(comms.registry.new_thread('native', frozenset(), str(tmp_path), session_file=str(source))) @@ -94,6 +97,8 @@ def test_claude_file_transport_reaches_the_same_cli_and_tool_answer(tmp_path, mo message=dict(content=[dict(type='tool_result', tool_use_id='read-1', content='contents')])) final = dict(type='assistant', uuid='final-record', timestamp='2026-10-16T12:00:03Z', message=dict(content=[dict(type='text', text='done')])) + final_position = len(''.join(json.dumps(row) + '\n' + for row in (prompt, call, result)).encode()) path.write_text(''.join(json.dumps(row) + '\n' for row in (prompt, call, result, final))) comms = prepared(tmp_path / 'wire') comms.registry.declare(comms.registry.new_thread('claude', frozenset(), str(tmp_path), execution=ExternalThreadExecution)) @@ -112,7 +117,8 @@ async def publish(): WatchSource.modified_file(item) for item in source.source_paths(comms.root)))) as changes: async for _ in changes: turns = await asyncio.to_thread(comms.transcripts.turns, 'claude', LastTurns()) - if any(event == AssistantTranscript('done', timestamp=datetime.fromisoformat(final['timestamp']).timestamp()) + if any(event == AssistantTranscript('done', position=final_position, + timestamp=datetime.fromisoformat(final['timestamp']).timestamp()) for turn in turns.turns for event in turn.events): break finally: diff --git a/tools/commander_page_journey.py b/tools/commander_page_journey.py new file mode 100644 index 0000000000..268c48af83 --- /dev/null +++ b/tools/commander_page_journey.py @@ -0,0 +1,205 @@ +"""Physical Toad journey on a scratch bus, using copies of real saved unit sessions. + +Run --prepare once, then run this file inside st on an isolated Xvfb display. +No provider is started, no synthetic transcript is written, and the original +sessions/registry are only read. Source and capture paths are recorded explicitly. +""" + +from __future__ import annotations + +import argparse +import asyncio +import os +import subprocess +import time +from dataclasses import replace +from pathlib import Path + +from textual.pilot import Pilot +from toad.app import ToadApp +from toad.navigation_target import FeedTarget +from toad.screens.commander import CommanderScreen, CommanderUnit +from toad.widgets.conversation import Conversation + +from agent_comms.active_route import LocalRoute +from agent_comms.change_report import FileLine +from agent_comms.child_process import NoProcess +from agent_comms.comms import Comms +from agent_comms.errors import ToolRefusedError +from agent_comms.goal_actions import CreateProblemGoalAction, SetGoalAction +from agent_comms.goals import CountCheckResult, NoGoal, OrderCheck, RequestGoalOrder +from agent_comms.progress_claims import WorkingClaim +from agent_comms.recorded_time import RecordedEventTime +from agent_comms.thread_status import StoppedThreadStatus +from agent_comms.todos import MechanicalOrderRun, OperationDelivery, PendingProblemOrder +from agent_comms.tool_caller import ThreadCaller +from agent_comms.tools import CommsOrderDoneTool, CommsSendTool, CommsTodoCreateTool + +COMMANDER = "claude-code" +UNITS = ("pi-cn-phase2", "pi-cn-phase5") + + +def prepare(output: Path, units: tuple[str, ...], ordered_at: float) -> None: + live = Comms(LocalRoute.default()) + root = output / "bus" + if root.exists(): + raise ValueError("Scratch bus already exists; preserve it and choose another output path.") + root.mkdir(parents=True) + comms = Comms(root) + source_notes = [] + for name in (COMMANDER, *units): + original = live.registry.require(name) + session = original.session_file + if session: + source = Path(session) + # Freeze a complete-line cut of the real saved session, not manufactured events. + with source.open("rb") as stream: + captured = stream.read(os.fstat(stream.fileno()).st_size) + copied = output / f"{name}.jsonl" + copied.write_bytes(captured.rsplit(b"\n", 1)[0] + b"\n") + copied.chmod(0o600) + session = str(copied) + source_notes.append(f"{name}: {source} -> {copied} ({copied.stat().st_size} bytes)") + comms.registry.declare( + replace(original, process_identity=NoProcess(), session_file=session, goal=NoGoal()), + StoppedThreadStatus(), + ) + comms.goals.update_goal( + COMMANDER, + SetGoalAction( + text="Verify the commander page on copied real unit sessions", + goal_id="page-request", + order=RequestGoalOrder(), + ), + ) + now = time.time() + for index, name in enumerate(units): + comms.goals.update_goal( + name, SetGoalAction(text="Copied unit for read-only page verification") + ) + order = PendingProblemOrder( + id=f"commander-page-{name}", + owner=FileLine(__file__, 1), + check=OrderCheck(str(output), "printf 1", CountCheckResult(1)), + edits=("Render the canonical order owner without copying its state",), + banned=("Do not launch a provider or change the original saved session",), + report="Report the actual visible commander page and its expanded unit card", + at=RecordedEventTime(ordered_at if not index else now), + first_call_deadline=RecordedEventTime(now + 120), + run=MechanicalOrderRun(), + ) + CommsTodoCreateTool( + owner=name, + order=CreateProblemGoalAction( + problem=f"P{index + 1}", + order=order, + request="page-request", + deadline=RecordedEventTime(now - 60 if not index else now + 1800), + commander=COMMANDER, + repo="OpenHCSDev/agent-comms", + text="Show the real order in the commander page", + ), + ).apply(comms, ThreadCaller(COMMANDER)) + try: + CommsSendTool( + sender=COMMANDER, + target="unregistered-page-target", + body="Verify that the commander can read an actual refused send.", + ).apply(comms, ThreadCaller(COMMANDER)) + except ToolRefusedError as error: + source_notes.append(f"Actual refused send: {error}") + else: + raise AssertionError("The deliberately absent target unexpectedly accepted a send.") + (output / "sources.txt").write_text("\n".join(source_notes) + "\n") + + +async def journey( + app: ToadApp, pilot: Pilot, output: Path, units: tuple[str, ...], momentum: bool +) -> None: + await pilot.pause(2) + # Open a real bus conversation and submit through its actual local-command boundary. + await app.selected_session.open_sidebar_target(FeedTarget()) + conversation = app.selected_session.query_one(Conversation) + await pilot.pause(1) + conversation.prompt.text = "/commander " + COMMANDER + conversation.prompt.focus() + await pilot.press("enter") + async with asyncio.timeout(90): + while not isinstance(app.screen, CommanderScreen) or len(app.screen.rows) != len(units): + await pilot.pause(0.2) + screen = app.screen + await pilot.pause(1) + + def capture(stage: str) -> None: + # The text is from the actual laid-out terminal strips, not the model. + text = "\n".join(strip.text for strip in screen._compositor.render_strips()) + (output / f"{stage}.txt").write_text(text + "\n") + subprocess.run(("import", "-window", "root", str(output / f"{stage}.png")), check=True) + + capture("level-0") + card = next(row for row in screen.query(CommanderUnit) if row.unit.name == units[0]) + if momentum: + assert "m/commit; next commit" in str(card.momentum_widget.content) + assert "completion unknown" in str(card.momentum_widget.content) + await pilot.click(card.query_one("CollapsibleTitle")) + async with asyncio.timeout(90): + while "Latest checkout commits:" not in str(card.detail_widget.content): + await pilot.pause(0.2) + await pilot.pause(1) + assert units[0] in str(card.title) + if len(units) > 1: + assert screen.rows[next(unit for unit in screen.rows if unit.name == units[1])].collapsed + capture("level-1") + + def deliver() -> None: + comms = Comms(output / "bus") + todo = comms.goals.todos.get("P1") + assert isinstance(todo.order, PendingProblemOrder) + CommsOrderDoneTool( + problem=todo.id, + order_id=todo.order.id, + delivery=OperationDelivery(), + claims=(WorkingClaim(str(output / "level-0.png")),), + ).apply(comms, ThreadCaller(COMMANDER)) + + await app.preparation.run_thread(deliver) + async with asyncio.timeout(90): + while "Delivered, waiting on install" not in str(card.title): + await pilot.pause(0.2) + await pilot.pause(0.5) + if momentum: + assert "order delivered" in str(card.momentum_widget.content) + capture("live-delivery") + # Verify collapse/return does not replace the underlying conversation or its draft. + await pilot.press("escape") + assert app.selected_session.query_one(Conversation) is conversation + (output / "result.txt").write_text( + "PASS: normal Toad, real scratch orders, level 0, expanded real turns, " + "and live delivery update without reopening.\n" + ) + app.exit() + + +def main() -> None: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--output", type=Path, required=True) + parser.add_argument("--prepare", action="store_true") + parser.add_argument("--units", nargs="+", default=UNITS) + parser.add_argument("--ordered-at", type=float, default=time.time() - 1200) + parser.add_argument("--momentum", action="store_true") + args = parser.parse_args() + args.output.mkdir(parents=True, exist_ok=True) + if args.prepare: + prepare(args.output, tuple(args.units), args.ordered_at) + return + os.environ.update(AGENT_COMMS_ROOT=str(args.output / "bus")) + app = ToadApp(project_dir=str(args.output)) + app.run( + auto_pilot=lambda pilot: journey(app, pilot, args.output, tuple(args.units), args.momentum) + ) + if app._exception: + raise app._exception + + +if __name__ == "__main__": + main()