Skip to content
Merged
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/channel_coding_tools.py
Original file line number Diff line number Diff line change
Expand Up @@ -69,7 +69,7 @@ def goal_permission(self) -> TurnGoalPermission: ...
async def reserve_room(self, call: CodingCall) -> RoomReservation:
registry = self.comms.registry
thread = self.thread
projection = StorageProjection(registry.storage_rule, thread)
projection = StorageProjection(registry.storage_rule, thread, self.comms)
effects = await asyncio.to_thread(projection.check_call, call)
work = tuple(dict.fromkeys(kind for effect in effects for kind in effect.work))
if work:
Expand Down
4 changes: 2 additions & 2 deletions src/agent_comms/cli_commands.py
Original file line number Diff line number Diff line change
Expand Up @@ -506,7 +506,7 @@ def apply(cls, command: PreparedCheckoutCliCommand, ctx: Comms, thread: Thread,
grant = command.grant(ctx, thread)
grant.require()
repository = command.source(ctx, grant)
projection = StorageProjection(ctx.registry.storage_rule, thread)
projection = StorageProjection(ctx.registry.storage_rule, thread, ctx)
if destination.exists():
CheckoutStorageEffect(destination, repository).check(projection)
Checkout(destination).prepare()
Expand All @@ -527,7 +527,7 @@ class RemoveCheckoutOperation(CheckoutOperation):
def apply(cls, command: PreparedCheckoutCliCommand, ctx: Comms, thread: Thread,
destination: Path) -> None:
RemoveCheckoutStorageEffect(destination).check(
StorageProjection(ctx.registry.storage_rule, thread))
StorageProjection(ctx.registry.storage_rule, thread, ctx))
Checkout(destination).retire()


Expand Down
8 changes: 8 additions & 0 deletions src/agent_comms/commander_frame_file.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,8 @@
from __future__ import annotations

from pathlib import Path

from agent_comms.service_passes import ServicePassStore
from agent_comms.locked_store import PatternFile, RuntimeFile


Expand All @@ -12,3 +15,8 @@
__slots__ = ()

filename = "commander_frames/{lookup}.json"

@classmethod
def source_paths(cls, root: Path) -> tuple[Path, ...]:
"""Frame readers hydrate service health too; both files wake their open view."""
return (*sorted(cls.matching_in(root)), ServicePassStore.path_in(root))
68 changes: 40 additions & 28 deletions src/agent_comms/commander_hud.py
Original file line number Diff line number Diff line change
@@ -1,3 +1,3 @@
"""Commander display derived from the native record owners in one read."""

from __future__ import annotations
Expand All @@ -5,7 +5,7 @@
import asyncio
import time
from abc import ABC, abstractmethod
from dataclasses import dataclass, field, replace
from dataclasses import dataclass, field
from pathlib import Path
from typing import TYPE_CHECKING, Annotated

Expand All @@ -18,15 +18,14 @@
from agent_comms.commander.vibe_check import OverloadDegradation
from agent_comms.commander_frame_file import CommanderFrameFile
from agent_comms.commander_service import CommanderService
from agent_comms.coordinator import Coordination
from agent_comms.errors import ToolRefusedError
from agent_comms.field_codec import DisclosureIdentity, OrderId, WireTransient, projected
from agent_comms.field_codec import DisclosureIdentity, OrderId, projected
from agent_comms.heartbeat_frame import HeartbeatFrame
from agent_comms.locked_store import JsonDocument
from agent_comms.plan_phases import Plan, PlanPhase
from agent_comms.order_story import OrderStory
from agent_comms.recorded_time import EventTime, RecordedEventTime, UnrecordedEventTime
from agent_comms.service_passes import ServicePassStore
from agent_comms.recorded_time import RecordedEventTime
from agent_comms.service_passes import BoardUnavailable, ServicePassStore, ServiceReading
from agent_comms.thread_identity import ThreadIncarnation
from agent_comms.todos import OrderProgress, ProgressPoint, ProgressRead, ProgressSource, PullRequest
from agent_comms.tool_result import ToolResult
Expand Down Expand Up @@ -256,7 +255,23 @@


@dataclass(frozen=True)
class CommanderHud(ToolResult):
class CommanderLoad(ServiceReading):
"""Retained load with the same recorded service evidence as its published frame."""

load: OverloadDegradation

def retained_status(self) -> str:
if self.services.heartbeat_current(self.heartbeat):
return 'Retained board rows and load:'
return 'STALE retained board rows and load:'

@projected(view='wire')
def header(self) -> tuple[str, ...]:
return (*self.service_status, self.retained_status(), str(self.load))


@dataclass(frozen=True)
class CommanderHud(CommanderLoad):
"""One disposable view of native records, shared by comms_hud and the commander board.

TodoStore owns orders, review timers and strikes; OrderEditCountRead owns edit counts;
Expand All @@ -268,14 +283,9 @@
commander: ThreadIncarnation
units: tuple[CommanderUnit, ...]
phases: tuple[CommanderPhase, ...]
load: OverloadDegradation
forks: tuple[ThreadIncarnation, ...]
usage: CodexUsage
board: tuple[BoardEntry, ...] = ()
heartbeat: Annotated[EventTime, WireTransient()] = field(
default_factory=UnrecordedEventTime, compare=False, kw_only=True,
)
"""Read-time projection of ServicePassStore's clock; never written into the frame."""

@classmethod
def read(cls, comms: Comms, commander: str) -> CommanderHud:
Expand Down Expand Up @@ -327,9 +337,9 @@
if snapshot.status(fork.name).receiving)
board = tuple(entry for packet in CommanderUpdate.latest(comms, owner)
for entry in packet.board(comms, commander))
return cls(owner, tuple(units), phases,
OverloadDegradation.from_orders(held, len(timers), sum(edits.values())),
forks, CodexUsage.read(), board)
return cls(commander=owner, units=tuple(units), phases=phases,
load=OverloadDegradation.from_orders(held, len(timers), sum(edits.values())),
forks=forks, usage=CodexUsage.read(), board=board)

def rows(self) -> tuple[HudRow, ...]:
"""Committed board order, actionable cards, live phases, then folded quiet units."""
Expand Down Expand Up @@ -370,8 +380,10 @@
@projected(view="wire")
def frame(self) -> tuple[str, ...]:
"""The load line, then one line per row in the order rows() declares."""
return (str(self.load), *self.heartbeat.aged_line("Tristan", time.time()),
*(row.line() for row in self.rows()))
return (*self.header, *(row.line() for row in self.rows()))

def load_reading(self) -> CommanderLoad:
return CommanderLoad(self.load, services=self.services, heartbeat=self.heartbeat)



Expand All @@ -386,26 +398,23 @@
"""

path: Path
reading: ServiceReading = field(default_factory=ServiceReading, compare=False, kw_only=True)

record_type = CommanderHud

def empty(self) -> CommanderHud:
raise ToolRefusedError(f"Tristan frame missing: {self.path}")
raise BoardUnavailable(self.reading, f'Tristan frame missing: {self.path}')

@classmethod
def current(cls, comms: Comms, commander: str) -> CommanderHud:
"""Read the heartbeat's publication, never repeat its forge and history work."""
owner = comms.registry.require(commander).incarnation.require_recorded()
passes = ServicePassStore.at(comms.root).read()
completed = passes.heartbeat_completed_at(owner)
if not completed.since(time.time() - 3 * Coordination.watch_period):
raise ToolRefusedError(
f"Tristan stale for {owner.name} since {completed.clock()}: {passes.stall()}"
)
hud = cls(cls.path_in(comms.root, lookup=str(owner.lookup))).read()
health = ServiceReading().with_services(passes, owner)
hud = cls(cls.path_in(comms.root, lookup=str(owner.lookup)), reading=health).read()
if hud.commander != owner:
raise ToolRefusedError(f"Tristan frame belongs to another incarnation of {owner.name}")
return replace(hud, heartbeat=completed)
raise BoardUnavailable(health, f'Tristan frame belongs to another incarnation of {owner.name}')
return hud.with_services(passes, owner)

@classmethod
def replace_all(cls, root: Path, huds: tuple[CommanderHud, ...]) -> None:
Expand All @@ -422,7 +431,10 @@
@classmethod
def read_all(cls, root: Path) -> tuple[CommanderHud, ...]:
passes = ServicePassStore.at(root).read()
return tuple(sorted((replace(hud, heartbeat=passes.heartbeat_completed_at(
hud.commander.require_recorded()))
for path in cls.matching_in(root) for hud in (cls(path).read(),)),
health = ServiceReading(services=passes)
huds = tuple(sorted((hud.with_services(passes, hud.commander.require_recorded())
for path in cls.matching_in(root) for hud in (cls(path, reading=health).read(),)),
key=lambda hud: hud.commander.name))
if not huds and passes.failing:
raise BoardUnavailable(health, 'no published commander frame.')
return huds
4 changes: 2 additions & 2 deletions src/agent_comms/heartbeat_frame.py
Original file line number Diff line number Diff line change
Expand Up @@ -13,7 +13,7 @@
from agent_comms.repo import Repo
from agent_comms.standing_rules import OwedAnswer, StandingRule
from agent_comms.thread_identity import ThreadIncarnation
from agent_comms.turn_digests import EditWorkplaces
from agent_comms.storage_rules import EditWorkplaces
from agent_comms.todos import (
ClosedHeadReading, HeadReviewRecord, ProgressRead, StoredClosedHead, StoredHeadReview,
)
Expand Down Expand Up @@ -76,5 +76,5 @@ def capture(cls, comms: Comms, commander: str, inputs: CommanderInputs,
OwedAnswer.read_scan(comms, commander, inputs),
closed,
reviews,
EditWorkplaces.capture(comms, snapshot, inputs.live, plan.todos, inputs.pulls),
EditWorkplaces.capture(comms, snapshot, inputs.live, inputs.pulls),
)
33 changes: 24 additions & 9 deletions src/agent_comms/order_tools.py
Original file line number Diff line number Diff line change
@@ -1,3 +1,3 @@
"""Todo and order tools, with the result records they return."""

from __future__ import annotations
Expand Down Expand Up @@ -33,11 +33,14 @@
JudgedCall,
ToolCallVerdict,
)
from .commander.vibe_check import OverloadDegradation
from .commander_hud import CommanderHud
from .commander_hud import CommanderHud, CommanderLoad
from .service_passes import ServicePassStore
from .comms import Comms
from .errors import ToolRefusedError
from .field_codec import FieldCodec, MessageId, OrderId, RecordProblems, WireName
from .problem_number import ProblemNumber
from .tool_declaration import EditCall
from .work_branches import WorkBranchError
from .goal_actions import CreateProblemGoalAction, OrderGoalAction
from .goals import Goal, OrderCheck
from .head_review import HeadReviewOrder, PreparedHeadReview, PullRequestHead
Expand Down Expand Up @@ -69,6 +72,7 @@
TurnView,
TurnsSince,
UnjudgedEdits,
UnavailableOrderEdits,
)


Expand Down Expand Up @@ -228,8 +232,17 @@
description = "Read original edit calls since each open order, excluding that order's recorded verdicts."

def apply(self, comms: Comms, caller: ToolCaller) -> UnjudgedEdits:
return UnjudgedEdits(tuple(edit for read in OrderEditsRead.for_commander(
comms, caller.subject(self.thread)) for edit in read.read()))
owner = comms.registry.require(caller.subject(self.thread)).incarnation.require_recorded()
passes = ServicePassStore.at(comms.root).read()
edits: list[EditCall] = []
unavailable: list[UnavailableOrderEdits] = []
for read in OrderEditsRead.for_commander(comms, owner.name):
try:
edits.extend(read.read())
except (WorkBranchError, OSError) as error:
unavailable.append(UnavailableOrderEdits(
ProblemNumber.parse(read.todo.id), read.thread.incarnation, error))
return UnjudgedEdits(tuple(edits), tuple(unavailable)).with_services(passes, owner)


@dataclass(frozen=True, kw_only=True)
Expand All @@ -244,18 +257,19 @@
@dataclass(frozen=True, kw_only=True)
class CommsCommanderLoadTool(BoardReadTool):
label = "Measured commander load"
description = "Read open orders, review timers and unjudged edits used by the overload rule."
description = "Read retained commander load with recorded service failures and its original completion clock."

def apply(self, comms: Comms, caller: ToolCaller) -> OverloadDegradation:
return CommanderHud.read(comms, caller.subject(self.thread)).load
def apply(self, comms: Comms, caller: ToolCaller) -> CommanderLoad:
return CommanderHud.read(comms, caller.subject(self.thread)).load_reading()


@dataclass(frozen=True, kw_only=True)
class CommsHudTool(ThreadReadTool):
label = "Commander HUD"
description = (
"Read the commander's latest published Tristan frame: orders, review timers, "
"strikes, edits, PRs, plan phases and load. A missing or stale Tristan frame is refused."
"strikes, edits, PRs, plan phases and load. Recorded failures precede retained data; "
"stale rows stay explicitly stale, and missing frames remain unavailable."
)

def apply(self, comms: Comms, caller: ToolCaller) -> CommanderHud:
Expand Down Expand Up @@ -325,7 +339,8 @@
label = "Read broken standing rules"
description = (
"Read the published Tristan board and HUD. Tristan owns "
"enforcement; this read never repeats it. A missing or stale Tristan frame is refused."
"enforcement; this read never repeats it. Recorded failures precede stale retained rows; "
"a missing frame remains explicitly unavailable."
)


Expand Down
4 changes: 2 additions & 2 deletions src/agent_comms/presentation.py
Original file line number Diff line number Diff line change
Expand Up @@ -895,7 +895,7 @@ def external_transcript_paths(views: HistoryViews) -> tuple[Path, ...]:


def commander_frame_paths(views: HistoryViews) -> tuple[Path, ...]:
return tuple(sorted(CommanderFrameFile.matching_in(views.root)))
return CommanderFrameFile.source_paths(views.root)


def expiry_period_changed(before: int, now: int) -> bool:
Expand Down Expand Up @@ -990,7 +990,7 @@ class WireRevision(ReadState):
# Activity-derived views change once per expiry period, not per tick.
expiry_tick: Annotated[
int,
WatchViews((ActivityView,)),
WatchViews((ActivityView, CommanderFramesView)),
Differs(expiry_period_changed),
]

Expand Down
59 changes: 51 additions & 8 deletions src/agent_comms/service_passes.py
Original file line number Diff line number Diff line change
@@ -1,3 +1,3 @@
"""Original completion times and current failures of Core's service watches."""

from __future__ import annotations
Expand All @@ -5,9 +5,13 @@
import time
from collections.abc import Iterable
from dataclasses import dataclass, field, replace
from typing import ClassVar
from typing import Annotated, ClassVar, Self

from .coordinator import Coordination
from .locked_store import DocumentChange, DurableFile, LockedStore
from .errors import ToolRefusedError
from .field_codec import WireTransient, projected
from .tool_result import ToolResult
from .recorded_time import EventTime, RecordedEventTime, UnrecordedEventTime
from .store_files import DependentStateTier, SingleFileStore
from .thread_identity import RecordedThreadIncarnation, ThreadIncarnation
Expand All @@ -20,32 +24,71 @@
since: float
error: str

def failure_line(self, watch: str) -> str:
return (f'FIX service pass {watch} failing since {RecordedEventTime(self.since).clock()}'
f' ({max(0, int(time.time() - self.since))}s elapsed):'
f' {self.error}. is anyone fixing it, or letting it pile up?'
' -> fix the owner of the failing read this wake')


@dataclass(frozen=True)
class ServicePasses:
failing: dict[str, FailingPass] = field(default_factory=dict)
commander_completed: dict[RecordedThreadIncarnation, float] = field(default_factory=dict)
"""Last observation completed for each commander, including quiet observations."""

def failure_lines(self) -> tuple[str, ...]:
return tuple(failure.failure_line(watch) for watch, failure in sorted(self.failing.items()))

def heartbeat_completed_at(self, commander: RecordedThreadIncarnation) -> EventTime:
"""The original completion clock, or its declared unrecorded state."""
if commander in self.commander_completed:
return RecordedEventTime(self.commander_completed[commander])
return UnrecordedEventTime()

def stall(self) -> str:
"""Why passes stopped completing: every recorded failure with its error, or, with none
recorded, that no watch is running at all."""
return "; ".join(
f"{watch} failing since {RecordedEventTime(failure.since).clock()}: {failure.error}"
for watch, failure in self.failing.items()
) or "no failure recorded, so no heartbeat is running"
def heartbeat_current(self, heartbeat: EventTime) -> bool:
"""Only the commander's completion clock determines heartbeat freshness."""
return heartbeat.since(time.time() - 3 * Coordination.watch_period)

def completed_since(self, commander: RecordedThreadIncarnation, since: float) -> bool:
"""A pass for this commander completed at or after ``since``."""
return self.heartbeat_completed_at(commander).since(since)


@dataclass(frozen=True)
class ServiceReading(ToolResult):
"""Read-time service evidence shared by board, load and independent edit replies."""

services: Annotated[ServicePasses, WireTransient()] = field(
default_factory=ServicePasses, compare=False, kw_only=True)
heartbeat: Annotated[EventTime, WireTransient()] = field(
default_factory=UnrecordedEventTime, compare=False, kw_only=True)

def with_services(self, passes: ServicePasses, owner: RecordedThreadIncarnation) -> Self:
return replace(self, services=passes, heartbeat=passes.heartbeat_completed_at(owner))

@projected(view='wire')
def service_status(self) -> tuple[str, ...]:
clock = self.heartbeat.aged_line('Tristan last completed', time.time())
failures = self.services.failure_lines()
if not failures and not self.services.heartbeat_current(self.heartbeat):
failures = ('no failure recorded, so no heartbeat is running',)
return (*failures, *(clock or ('Tristan last completed: unrecorded',)))

def require_current(self) -> None:
"""A disclosure without a health header cannot serve stale cards silently."""
if not self.services.heartbeat_current(self.heartbeat):
raise BoardUnavailable(self, 'the commander heartbeat is stale')


class BoardUnavailable(ToolRefusedError):
"""A missing or mismatched publication retains the service evidence that explains it."""

def __init__(self, reading: ServiceReading, reason: str) -> None:
self.reading = reading
super().__init__('\n'.join((*reading.service_status, f'Board unavailable: {reason}')))


@dataclass(frozen=True, slots=True)
class ServicePassStore(SingleFileStore, DurableFile, LockedStore[ServicePasses], DependentStateTier):
"""The connector's log reaches nobody; the commander's standing rules read this file."""
Expand Down
Loading
Loading