diff --git a/src/agent_comms/commander/judge.py b/src/agent_comms/commander/judge.py index 75b39b9ea3..a0a7425be0 100644 --- a/src/agent_comms/commander/judge.py +++ b/src/agent_comms/commander/judge.py @@ -16,6 +16,8 @@ the commander to that unit. Nothing here judges code; it only refuses to let an edit pass unread. """ +from __future__ import annotations + import hashlib import json import os @@ -24,12 +26,16 @@ import time from typing import Annotated from abc import abstractmethod -from dataclasses import asdict, dataclass, field +from dataclasses import asdict, dataclass +from functools import cached_property from agent_comms.comms import Comms from agent_comms.change_report import FileLine from agent_comms.field_codec import Argument, FieldCodec +from agent_comms.declared_family import DeclaredFamily +from agent_comms.head_review import PullRequestHead +from agent_comms.repo import Repo from agent_comms.todos import ( AcceptedVerdict, BluntText, @@ -37,6 +43,9 @@ DroppedVerdict, PeerAcceptedVerdict, ReviewedHeadVerdict, + RejectedHeadVerdict, + HeadVerdict, + HeadReviewOrder, SendBackVerdict, StrikeReason, SupersededVerdict, @@ -145,26 +154,92 @@ def run(self, ctx: order.Commander) -> None: alarm(ctx) -@dataclass(frozen=True, kw_only=True) -class HeadVerb(JudgeVerb): - help = ("accept an order's reviewed PR head: every edit before it is judged and " - "agent-comms/review is posted on the commit") - problem: Annotated[str, Argument(flags=("problem",), options={"metavar": "PROBLEM"})] - head: Annotated[str, Argument(flags=("head",), options={"metavar": "SHA"})] - serves: Annotated[ - str, Argument(options={"help": "exact words of the order the head carries out"}) - ] +class ReviewSubject(DeclaredFamily): + """Decode the CLI's mutually exclusive selectors once, then keep the selected subject.""" - def run(self, ctx: order.Commander) -> None: - held = [todo for todo in ctx.held_orders() if todo.id == self.problem] - if not held: - sys.exit(f'Refused: {self.problem} is not an order this commander holds') - todo, = held - verdict = ReviewedHeadVerdict(todo.revision, ctx.actor_identity, todo.held(), - quote=self.serves, head=self.head) - ctx.call(CommsHeadVerdictTool(problem=todo.id, operation=verdict)) + @classmethod + def from_options(cls, options: HeadVerdictVerb, ctx: order.Commander) -> ReviewSubject: + matches = tuple(member for member in cls.members_with(cls) if member.selected(options)) + if len(matches) != 1: + sys.exit('Refused: select either edit IDs or --pr') + member, = matches + return member.decode_options(options, ctx) + + @classmethod + @abstractmethod + def selected(cls, options: HeadVerdictVerb) -> bool: ... + + @classmethod + @abstractmethod + def decode_options(cls, options: HeadVerdictVerb, ctx: order.Commander) -> ReviewSubject: ... + + @abstractmethod + def run(self, options: VerdictVerb, ctx: order.Commander) -> None: ... + + +@dataclass(frozen=True) +class EditReview(ReviewSubject): + ids: tuple[str, ...] + + @classmethod + def selected(cls, options: HeadVerdictVerb) -> bool: + return bool(options.ids) + + @classmethod + def decode_options(cls, options: HeadVerdictVerb, ctx: order.Commander) -> EditReview: + if options.head or options.repo: + sys.exit('Refused: --head and --repo belong to --pr, not edit IDs') + return cls(options.ids) + + def run(self, options: VerdictVerb, ctx: order.Commander) -> None: + if not self.ids: + sys.exit('Refused: select at least one edit ID:CODE') + orders = {todo.id: todo for todo in ctx.held_orders() + if not options.problem or todo.id == options.problem} + edits = {edit.id: edit for edit in Edit.unjudged(ctx.comms, ctx.commander) + if edit.problem in orders} + ids = [token.split(':')[0] for token in self.ids] + if missing := [i for i in ids if i not in edits]: + sys.exit(f'Refused: unknown edit ids {missing}') + if unread := [token for token in self.ids + if token != f'{token.split(":")[0]}:{edits[token.split(":")[0]].code()}']: + sys.exit(f'Refused: wrong or missing read codes {unread}; read each edit to its "=== end" line') + for i in ids: + if not any(line in edits[i].text() for line in options.lines): + sys.exit(f'Refused: none of your --line quotes is in edit {i}; judge it by its own text') + selected = {i: orders[edits[i].problem] for i in ids} + options.check_edits([edits[i] for i in ids]) + for i, todo in selected.items(): + options.operation(ctx, todo, edits[i]).prepare(ctx.comms.goals, todo).apply(todo) + current = {todo.id: todo for todo in selected.values()} + for i, todo in selected.items(): + result = ctx.call(CommsVerdictTool(problem=todo.id, + operation=options.operation(ctx, current[todo.id], edits[i]))) + current[todo.id] = result.todo left = len(Edit.unjudged(ctx.comms, ctx.commander)) - print(f'head {self.head} accepted on {todo.id}; {left} still unjudged') + print(f'{options.declared_name}: {len(ids)} edits; {left} still unjudged') + + +@dataclass(frozen=True) +class PullRequestReview(ReviewSubject): + request: PullRequestHead + + @classmethod + def selected(cls, options: HeadVerdictVerb) -> bool: + return bool(options.pr) + + @classmethod + def decode_options(cls, options: HeadVerdictVerb, ctx: order.Commander) -> PullRequestReview: + return cls(PullRequestHead(Repo.selected(ctx.repos, options.repo).github, + options.pr, options.head)) + + def run(self, options: HeadVerdictVerb, ctx: order.Commander) -> None: + options.check_edits([]) + linked = HeadReviewOrder.for_request(ctx.held_orders(), self.request, options.serves, + options.head_evidence(), options.problem) + ctx.call(CommsHeadVerdictTool(request=self.request, + operation=options.head_operation(), order=linked)) + print(f'PR #{self.request.pr} head {self.request.head}: {options.declared_name}') @dataclass(frozen=True, kw_only=True) @@ -174,8 +249,8 @@ class VerdictVerb(JudgeVerb): ids: Annotated[ tuple[str, ...], - Argument(flags=("ids",), options={"nargs": "+", "metavar": "ID:CODE"}), - ] + Argument(flags=("ids",), options={"nargs": "*", "metavar": "ID:CODE"}), + ] = () why: str problem: Annotated[ str, @@ -191,7 +266,7 @@ class VerdictVerb(JudgeVerb): "help": "an exact diff line; repeat for each", }, ), - ] + ] = () @abstractmethod def operation(self, ctx: order.Commander, todo: Todo, edit: Edit) -> ToolCallVerdict: ... @@ -199,42 +274,38 @@ def operation(self, ctx: order.Commander, todo: Todo, edit: Edit) -> ToolCallVer @abstractmethod def check_edits(self, edits: list[Edit]) -> None: ... + def subject(self, ctx: order.Commander) -> ReviewSubject: + return EditReview(self.ids) + def run(self, ctx: order.Commander) -> None: if len(self.why.split()) < 6: sys.exit('Refused: state which intent the edit serves or violates, and why (6+ words).') - orders = {todo.id: todo for todo in ctx.held_orders() - if not self.problem or todo.id == self.problem} - edits = {edit.id: edit for edit in Edit.unjudged(ctx.comms, ctx.commander) - if edit.problem in orders} - ids = [token.split(':')[0] for token in self.ids] - if missing := [i for i in ids if i not in edits]: - sys.exit(f'Refused: unknown edit ids {missing}') - if unread := [token for token in self.ids - if token != f'{token.split(":")[0]}:{edits[token.split(":")[0]].code()}']: - sys.exit(f'Refused: wrong or missing read codes {unread}; read each edit to its "=== end" line') - for i in ids: - if not any(line in edits[i].text() for line in self.lines): - sys.exit(f'Refused: none of your --line quotes is in edit {i}; judge it by its own text') - selected = {i: orders[edits[i].problem] for i in ids} - self.check_edits([edits[i] for i in ids]) - # Validate the whole batch through the operation owner before its first write. - for i, todo in selected.items(): - self.operation(ctx, todo, edits[i]).prepare(ctx.comms.goals, todo).apply(todo) - current = {todo.id: todo for todo in selected.values()} - for i, todo in selected.items(): - result = ctx.call(CommsVerdictTool(problem=todo.id, - operation=self.operation(ctx, current[todo.id], edits[i]))) - current[todo.id] = result.todo - left = len(Edit.unjudged(ctx.comms, ctx.commander)) - print(f'{self.declared_name}: {len(ids)} edits; {left} still unjudged') + self.subject(ctx).run(self, ctx) @dataclass(frozen=True, kw_only=True) -class OkVerb(VerdictVerb): - help = "the edit serves the order" - serves: Annotated[ - str, Argument(options={"help": "exact words of the order"}) - ] +class HeadVerdictVerb(VerdictVerb, declared_name=None): + pr: Annotated[int, Argument(options={"help": "review this PR instead of individual edits"})] = 0 + head: Annotated[str, Argument(options={"help": "the exact reviewed SHA; required with --pr"})] = "" + repo: Annotated[str, Argument(options={"help": "registered repository name; optional for one repo"})] = "" + serves: Annotated[str, Argument(options={"help": "exact words of the linked order"})] = "" + + def subject(self, ctx: order.Commander) -> ReviewSubject: + return ReviewSubject.from_options(self, ctx) + + def head_evidence(self) -> tuple[FileLine, ...]: + return () + + @abstractmethod + def head_operation(self) -> HeadVerdict: ... + + +@dataclass(frozen=True, kw_only=True) +class OkVerb(HeadVerdictVerb): + help = "the edit or exact PR head is accepted" + + def head_operation(self) -> HeadVerdict: + return ReviewedHeadVerdict() no_owner: Annotated[ str, Argument( @@ -296,13 +367,8 @@ def check_edits(self, edits: list[Edit]) -> None: pass @dataclass(frozen=True, kw_only=True) -class BackVerb(VerdictVerb): - help = "send the edit back; Core publishes the recorded reason once" - # A bare annotation would inherit VerdictVerb's default; field() keeps it required. - problem: Annotated[ - str, - Argument(options={"metavar": "PROBLEM", "help": "required problem id from order board"}), - ] = field() +class BackVerb(HeadVerdictVerb): + help = "reject the edit or exact PR head; Core records its reason once" roast: Annotated[ str, Argument(options={"help": "blunt, why it is stupid, with because"}) ] @@ -326,28 +392,38 @@ class BackVerb(VerdictVerb): str, Argument(options={"help": "PATH:LINE of the declaration that owns it"}) ] + @cached_property def slop(self) -> Slop: return FieldCodec.decode(Slop, dict(pattern=self.pattern, fact=self.fact, at=self.at, derive=self.derive, by=self.by)) - def operation(self, ctx: order.Commander, todo: Todo, edit: Edit) -> SendBackVerdict: - return SendBackVerdict(todo.revision, ctx.actor_identity, - todo.held(), edit.id, - reason=StrikeReason("\n".join(( - self.why, self.roast, - *(f" {line.strip()}" for line in self.lines), - self.slop().render(), f"What replaces it: {self.fix}", - ))), - evidence=(FileLine.from_wire(self.at),)) - - def check_edits(self, edits: list[Edit]) -> None: + @cached_property + def rejection_reason(self) -> StrikeReason: if len(self.roast.split()) < 12 or not re.search(r'\bbecause\b', self.roast): sys.exit('Refused: --roast needs 12+ words with because, naming the duplicated fact and its owner') if len(self.fix.split()) < 8: sys.exit('Refused: --fix needs 8+ words naming the owner and replacing shape') - self.slop() order.refuse(self.roast, BluntText) order.refuse(' '.join((self.why, self.roast, self.fix)), *DirectiveText.checks()) + return StrikeReason("\n".join(( + self.why, self.roast, + *(f" {line.strip()}" for line in self.lines), + self.slop.render(), f"What replaces it: {self.fix}", + ))) + + def head_evidence(self) -> tuple[FileLine, ...]: + return (FileLine.from_wire(self.at),) + + def head_operation(self) -> HeadVerdict: + return RejectedHeadVerdict(self.rejection_reason) + + def operation(self, ctx: order.Commander, todo: Todo, edit: Edit) -> SendBackVerdict: + return SendBackVerdict(todo.revision, ctx.actor_identity, + todo.held(), edit.id, reason=self.rejection_reason, + evidence=self.head_evidence()) + + def check_edits(self, edits: list[Edit]) -> None: + _ = self.rejection_reason def main(args, ctx): JudgeVerb.parse(args or ['show']).apply(ctx) diff --git a/src/agent_comms/commander/standing.py b/src/agent_comms/commander/standing.py index 782739986c..bf25a87e88 100644 --- a/src/agent_comms/commander/standing.py +++ b/src/agent_comms/commander/standing.py @@ -24,6 +24,8 @@ from agent_comms.read_basis import RepliesDisplayScope from agent_comms.recorded_time import RecordedEventTime from agent_comms.registry_document import RegistrySnapshot +from agent_comms.repository_name import RepositoryName +from agent_comms.service_passes import ServicePassStore from agent_comms.standing_rules import ( DatedStandingRule, IndependentStandingRule, @@ -51,7 +53,11 @@ class InstallDue(IndependentStandingRule, UndatedStandingRule): one install each. The window is read from the merges' own git times, not a schedule. """ - why: str + repo: RepositoryName + main: str + installed: str + commits: int = field(compare=False) + oldest_minutes: int = field(compare=False) settle: ClassVar[float] = 30 * 60 @classmethod @@ -70,14 +76,15 @@ def violations(cls, comms: Comms, commander: str) -> tuple[Self, ...]: f"{recorded.commit}..{main}").split() waited = now - min(map(int, merged), default=now - cls.settle) if waited >= cls.settle: - due.append(cls( - f"{repo.github.to_wire()}: main {main}, installed {recorded.commit}, " - f"{len(merged)} commits, oldest {int(waited // 60)} min")) + due.append(cls(repo.github, main, recorded.commit, + len(merged), int(waited // 60))) return tuple(due) def line(self) -> str: - return (f"INSTALL DUE ({self.why}) -> the installer rehearses and installs main with " - "agent-comms install and posts the result to the commander") + return (f"INSTALL DUE ({self.repo.to_wire()}: main {self.main}, installed " + f"{self.installed}, {self.commits} commits, oldest {self.oldest_minutes} min) " + "-> the installer rehearses and installs main with agent-comms install " + "and posts the result to the commander") @dataclass(frozen=True) @@ -244,17 +251,17 @@ def idle(cls, comms: Comms, commander: str, scan: StandingScan) -> list[Thread]: if TurnDigestRead(comms.root, candidate, bound).read().turns), key=lambda thread: thread.name) - def _identity(self) -> tuple[object, ...]: + def comparison_values(self) -> tuple[object, ...]: # Observation times and proposed idle units change; the row and its disposition do not. - return type(self), self.todo.id + return (self.todo.id,) def __eq__(self, other: object) -> bool: if not isinstance(other, UnownedWheelSpin): return NotImplemented - return self._identity() == other._identity() + return type(self) is type(other) and self.comparison_values() == other.comparison_values() def __hash__(self) -> int: - return hash(self._identity()) + return hash((type(self), self.comparison_values())) @classmethod def violations(cls, comms: Comms, commander: str) -> tuple[UnownedWheelSpin, ...]: @@ -321,8 +328,8 @@ def line(self) -> str: class CheckoutHeldWheelSpin(UnownedWheelSpin, declared_name=None): work: OrderCheckoutWork - def _identity(self) -> tuple[object, ...]: - return (*super()._identity(), self.work.owner.incarnation, self.work.checkout) + def comparison_values(self) -> tuple[object, ...]: + return (*super().comparison_values(), self.work.owner.incarnation, self.work.checkout) def line(self) -> str: return (f"WORK HELD: {self.todo.id} {self.work.owner.name} retains work in " @@ -360,38 +367,32 @@ def pass_body(self, comms: Comms, observed: tuple[UnownedObservation, ...]) -> N class WaitingUnownedWheelSpin(UnownedWheelSpin, declared_name=None): """A row needs a new unit; a proposed dispatch name is not an idle thread.""" - def heartbeat_group(self) -> object: - return type(self) - - def heartbeat_line(self, current: Sequence[WaitingUnownedWheelSpin]) -> str: - rows = ", ".join(sorted(rule.todo.id for rule in current)) - return f"{len(current)} unowned rows wait for a new unit: {rows}" - @dataclass(frozen=True) class HeartbeatSilent(IndependentStandingRule, DatedStandingRule, declared_name=None): - """Missing or stale heartbeat service for the commander's current open requests. - - The heartbeat beats on every pass, so its silence is the failure; this rule reads the - bus in the commander's own tool process, where a dead heartbeat cannot hide it. - """ + """A missing or stale successful pass, independent of whether a DM needed sending.""" seconds: ClassVar[int] = 180 since: float @classmethod - def violations(cls, comms: Comms, commander: str) -> tuple[Self, ...]: - return tuple(rule for member in cls.members_with(cls) - for rule in member.violations(comms, commander)) - - @staticmethod - def beats(comms: Comms, commander: str) -> tuple[Message, ...]: - snapshot = comms.registry.snapshot() - if snapshot.canonical_name(CommanderHeartbeat.sender) not in snapshot.threads: - return () - return comms.bus.bus.latest_addressed( - snapshot.names_for(CommanderHeartbeat.sender), snapshot.names_for(commander), - ) + def scan_owner(cls) -> type[StandingRule]: + return HeartbeatSilent + + @classmethod + def violations(cls, comms: Comms, commander: str) -> tuple[HeartbeatSilent, ...]: + owner = comms.registry.require(commander) + completed = ServicePassStore.at(comms.root).read().commander_completed.get(owner.incarnation) + now = time.time() + found = [] + for request in CommanderHeartbeat.requested_by(owner, comms.goals): + if completed is None or completed < request.created_at: + rule = HeartbeatNotStarted(request.created_at + cls.seconds, request.id) + else: + rule = MissedHeartbeat(completed + cls.seconds) + if rule.since < now: + found.append(rule) + return tuple(found) def began(self) -> float: return self.since @@ -399,22 +400,9 @@ def began(self) -> float: @dataclass(frozen=True) class MissedHeartbeat(HeartbeatSilent): - beat: int - - @classmethod - def violations(cls, comms: Comms, commander: str) -> tuple[Self, ...]: - requests = CommanderHeartbeat.requested_by(comms.registry.require(commander), comms.goals) - if not requests: - return () - now = time.time() - found = tuple( - cls(since=max(request.created_at, beat.timestamp) + cls.seconds, beat=beat.seq) - for request in requests for beat in cls.beats(comms, commander) - ) - return tuple(rule for rule in found if rule.since < now) - def line(self) -> str: - return (f"HEARTBEAT SILENT: no beat since seq {self.beat}, silent past " + return (f"HEARTBEAT SILENT: no completed pass since " + f"{RecordedEventTime(self.since - self.seconds).clock()}, silent past " f"{RecordedEventTime(self.since).clock()}" " -> read the connector journal for the failing pass and fix its owner") @@ -423,16 +411,6 @@ def line(self) -> str: class HeartbeatNotStarted(HeartbeatSilent): request: OwnerRequestId - @classmethod - def violations(cls, comms: Comms, commander: str) -> tuple[Self, ...]: - requests = CommanderHeartbeat.requested_by(comms.registry.require(commander), comms.goals) - if not requests or cls.beats(comms, commander): - return () - now = time.time() - found = tuple(cls(since=request.created_at + cls.seconds, request=request.id) - for request in requests) - return tuple(rule for rule in found if rule.since < now) - def line(self) -> str: return (f"HEARTBEAT NOT STARTED: no beat for request {self.request.value}, overdue since " f"{RecordedEventTime(self.since).clock()}" @@ -441,8 +419,8 @@ def line(self) -> str: @dataclass(frozen=True) class Unjudged(ReasonedStandingRule, UndatedStandingRule): limit: ClassVar[int] = 30 - text = ("UNJUDGED {why} edits -> accept each reviewed PR head with judge head (one verdict" - " judges every edit before it), or fork judges for the rest") + text = ("UNJUDGED {why} edits -> accept each reviewed PR head with judge ok --pr N --head SHA" + " (one verdict judges every edit before it), or fork judges for the rest") @classmethod def reason(cls, comms: Comms, commander: str) -> str | None: diff --git a/src/agent_comms/commander/vibe_check.py b/src/agent_comms/commander/vibe_check.py index d5e6bbce6c..fc4cf86dd1 100644 --- a/src/agent_comms/commander/vibe_check.py +++ b/src/agent_comms/commander/vibe_check.py @@ -210,13 +210,6 @@ def scan_observed(cls, comms: Comms, commander: str, for sign in member.found(comms, commander, frame) ) - def heartbeat_group(self) -> object: - return VibeCheck - - def heartbeat_line(self, current: Sequence[Self]) -> str: - names = ", ".join(sorted({check.sign.signal.declared_name for check in current})) - return f"VIBE CHECK ({names}) -> answer the {VibeCheckSender.sender} messages now" - def enforce(self, comms: Comms, commander: str) -> None: indictment = self.indictment() VibeCheckSender.burst_once(comms, commander, indictment, indictment[0]) @@ -234,7 +227,8 @@ def indictment(self) -> tuple[str, ...]: ) def line(self) -> str: - return self.heartbeat_line((self,)) + return (f"VIBE CHECK ({self.sign.signal.declared_name}) " + f"-> answer the {VibeCheckSender.sender} messages now") @dataclass(frozen=True) diff --git a/src/agent_comms/commander_heartbeat.py b/src/agent_comms/commander_heartbeat.py index 620c0c9e6e..9edeb96dca 100644 --- a/src/agent_comms/commander_heartbeat.py +++ b/src/agent_comms/commander_heartbeat.py @@ -2,14 +2,15 @@ from __future__ import annotations -from dataclasses import dataclass, field +from dataclasses import dataclass from typing import TYPE_CHECKING +from agent_comms.commander_pings import CommanderUpdate, RulePing, UnitPing from agent_comms.commander_service import CommanderService from agent_comms.heartbeat_frame import HeartbeatFrame from agent_comms.message_bus import WatchPass +from agent_comms.service_passes import ServicePassStore from agent_comms.standing_rules import StandingRule -from agent_comms.thread_identity import ThreadIncarnation from agent_comms.threads import Thread from agent_comms.todos import ProgressRead from agent_comms.ui_model.unit_state import UnitState @@ -19,46 +20,6 @@ from agent_comms.todos import Todo -@dataclass -class CommanderReview: - """One commander's handled observations and its last changed beat. - - Seq 0 precedes every published message: no changed beat has been sent yet. - """ - - rules: frozenset[StandingRule] - states: frozenset[UnitState] = frozenset() - changed: int = 0 - - def wake(self, comms: Comms, sender: str, commander: Thread, - rules: tuple[StandingRule, ...], states: tuple[UnitState, ...]) -> None: - """Every pass beats, so silence is a failure; an unchanged beat is one short line.""" - found_rules, found_states = frozenset(rules), frozenset(states) - actionable = (*rules, *(state for state in states if state.attention())) - old_rules = self.rules - previous = self.rules | self.states - if found_rules == self.rules and found_states == self.states and self.changed: - comms.messaging.send(sender, commander.name, - f"COMMANDER HEARTBEAT: unchanged since seq {self.changed}," - f" {len(actionable)} violations.") - return - # Mark the observation before any effect; an uncertain publication is not replayed. - self.rules, self.states = found_rules, found_states - rejected = StandingRule.enforce_all( - comms, commander.name, (rule for rule in actionable if rule not in previous) - ) - # A checked rejection has no effects and must not suppress a fresh observation. - self.rules -= rejected - self.states -= rejected - lines = ["COMMANDER HEARTBEAT: unit states and standing rules changed."] - lines.extend(state.line() for state in states) - lines.extend(StandingRule.heartbeat_changes(old_rules, self.rules)) - lines.extend(rule.rejected_line() for rule in actionable if rule in rejected) - if not actionable: - lines.append("Standing rules clear.") - self.changed = comms.messaging.send(sender, commander.name, "\n".join(lines)).message.seq - - @dataclass(frozen=True) class CommanderObservation: thread: Thread @@ -72,15 +33,15 @@ def read(cls, comms: Comms, thread: Thread, """One frame's progress facts come from ``reading``; Core reads the live forge and Git.""" frame = HeartbeatFrame.read(comms, thread.name, reading()) pulls = frame.pulls + for repo in comms.registry.snapshot().commander_project.repository_names(): + pulls.open(repo) states = UnitState.current(comms, thread.name, frame) rules = StandingRule.broken(comms, thread.name, observed=states, frame=frame) merged = tuple(todo for state in states for todo in state.merged_orders(comms, pulls)) return cls(thread, merged, rules, states) -@dataclass class CommanderHeartbeat(CommanderService, WatchPass[tuple[CommanderObservation, ...]]): - reviews: dict[ThreadIncarnation, CommanderReview] = field(default_factory=dict) def observe(self, comms: Comms) -> tuple[CommanderObservation, ...]: return tuple( @@ -89,13 +50,30 @@ def observe(self, comms: Comms) -> tuple[CommanderObservation, ...]: ) def pass_body(self, comms: Comms, observed: tuple[CommanderObservation, ...]) -> None: - """Inactive commanders retire their review after the pass enters the bus.""" + """The committed packet survives connector replacement; local memory owns nothing.""" self.register(comms.registry) - live = {item.thread.incarnation for item in observed} - self.reviews = { - owner: review for owner, review in self.reviews.items() if owner in live - } - for item in observed: - comms.goals.todos.record_merges(item.merged) - review = self.reviews.setdefault(item.thread.incarnation, CommanderReview(frozenset())) - review.wake(comms, self.sender, item.thread, item.rules, item.states) + completed = [] + try: + for item in observed: + comms.goals.todos.record_merges(item.merged) + previous = CommanderUpdate.latest(comms, item.thread.incarnation) + known = frozenset(previous.handled) if previous else frozenset() + rejected = StandingRule.enforce_all( + comms, item.thread.name, + (*(rule for rule in item.rules if RulePing.capture(rule) not in known), + *(state for state in item.states + if state.attention() and UnitPing.capture(state) not in known)), + ) + update = CommanderUpdate.derive( + item.thread.incarnation, + tuple(state for state in item.states if state not in rejected), + tuple(rule for rule in item.rules if rule not in rejected), previous, + ) + if update is not None: + update.send(comms, previous) + completed.append(item.thread.incarnation) + finally: + # A later commander's failure does not erase passes already completed. + ServicePassStore.at(comms.root).heartbeat_passed( + completed, (item.thread.incarnation for item in observed), + ) diff --git a/src/agent_comms/commander_pings.py b/src/agent_comms/commander_pings.py new file mode 100644 index 0000000000..912f71cced --- /dev/null +++ b/src/agent_comms/commander_pings.py @@ -0,0 +1,179 @@ +"""Committed commander notifications own what the heartbeat has already handled. + +Quiet state changes are passive notices. Urgent packets lead with five decisions +and publish every current observation; no process-local cache owns them. +""" +from __future__ import annotations + +from abc import abstractmethod +from dataclasses import dataclass, field +from typing import TYPE_CHECKING, ClassVar, Self + +from agent_comms.declared_family import DeclaredFamily +from agent_comms.commander_service import CommanderService +from agent_comms.errors import RelationViolationError +from agent_comms.message_reference import MessageReference +from agent_comms.messages import ( + ConversationRole, InfoKind, Message, MessageRole, NoticeRole, +) +from agent_comms.standing_rules import StandingRule +from agent_comms.task_sources import TaskAttachment +from agent_comms.text_digest import TextDigest +from agent_comms.thread_identity import ThreadIncarnation +from agent_comms.ui_model.unit_state import UnitState + +if TYPE_CHECKING: + from agent_comms.bus import BusReader + from agent_comms.comms import Comms + from agent_comms.registry_document import RegistrySnapshot + + +@dataclass(frozen=True) +class OrderKey: + problem: str + order: str + + +@dataclass(frozen=True) +class Ping(DeclaredFamily, affix="Ping"): + text: str = field(compare=False, kw_only=True) + + @property + @abstractmethod + def key(self) -> object: ... + + +@dataclass(frozen=True) +class RulePing(Ping): + observation: TextDigest + + @property + def key(self) -> TextDigest: + return self.observation + + @classmethod + def capture(cls, rule: StandingRule) -> Self: + return cls(rule.observation_key(), text=rule.line()) + + +@dataclass(frozen=True) +class UnitPing(Ping): + """One handled observation, not a separately writable current unit state.""" + unit: ThreadIncarnation + state: type[UnitState] + orders: tuple[OrderKey, ...] + questions: tuple[MessageReference, ...] + observation: TextDigest + + @property + def key(self) -> ThreadIncarnation: + return self.unit + + @classmethod + def capture(cls, state: UnitState) -> Self: + return cls( + state.unit, type(state), + tuple(OrderKey(todo.id, todo.order.checked().id) + for todo in state.source.todos if todo.id in state.orders), + tuple(message.reference for message in state.owed), + state.observation_key(), text=state.line(), + ) + + +@dataclass(frozen=True) +class CommanderUpdate(TaskAttachment, declared_name=None): + commander: ThreadIncarnation + handled: tuple[Ping, ...] + role: ClassVar[MessageRole] + + @abstractmethod + def body(self, previous: CommanderUpdate | None) -> str: ... + + def changed_pings(self, previous: CommanderUpdate | None, + presented: tuple[Ping, ...] = ()) -> tuple[Ping, ...]: + known = frozenset(previous.handled) if previous else frozenset() + return tuple(ping for ping in self.handled if ping not in known and ping not in presented) + + def changed_lines(self, previous: CommanderUpdate | None, + presented: tuple[Ping, ...] = ()) -> tuple[str, ...]: + prior = previous.handled if previous else () + current = frozenset(ping.key for ping in self.handled) + return ( + *(ping.text for ping in self.changed_pings(previous, presented)), + *(f"NO LONGER CURRENT: {ping.text}" for ping in prior if ping.key not in current), + ) + + def require_publication(self, sender: str, registry: RegistrySnapshot, + original_source: BusReader) -> None: + if registry.require(sender).incarnation != registry.require(CommanderService.sender).incarnation: + raise RelationViolationError("Only the commander heartbeat publishes its observations") + + def require_target(self, target: str, registry: RegistrySnapshot) -> None: + if registry.require(target).incarnation != self.commander: + raise RelationViolationError("Commander observations changed recipient incarnation") + + @classmethod + def latest(cls, comms: Comms, commander: ThreadIncarnation) -> CommanderUpdate | None: + snapshot = comms.registry.snapshot() + messages = comms.bus.bus.latest_addressed( + snapshot.names_for(CommanderService.sender), snapshot.names_for(commander.name), + ) + if not messages: + return None + message, = messages + # Ordinary messages carry no commander-observation evidence. + update = message.task + if not isinstance(update, CommanderUpdate): + return None + return update if update.commander == commander else None + + @classmethod + def derive(cls, commander: ThreadIncarnation, states: tuple[UnitState, ...], + rules: tuple[StandingRule, ...], + previous: CommanderUpdate | None) -> CommanderUpdate | None: + known = {ping.key: ping for ping in previous.handled} if previous else {} + quiet = tuple(UnitPing.capture(state) for state in states if not state.attention()) + urgent = (*(UnitPing.capture(state) for state in states if state.attention()), + *(RulePing.capture(rule) for rule in rules)) + decisions = tuple(ping for ping in urgent if known.get(ping.key) != ping)[:5] + handled = (*quiet, *urgent) + if decisions: + return CommanderDecisions(commander, handled, decisions) + if previous is None or handled != previous.handled: + return CommanderNotice(commander, handled) + return None + + def send(self, comms: Comms, previous: CommanderUpdate | None) -> Message: + from agent_comms.bus_publication import AgentOrigin + from agent_comms.channel_targets import ThreadTarget + from agent_comms.commander_heartbeat import CommanderHeartbeat + + message = Message( + sender=CommanderHeartbeat.sender, target=self.commander.name, + body=self.body(previous), type=InfoKind, role=self.role, task=self, + ) + return comms.bus.publisher.publish_ordinary( + message, origin=AgentOrigin(), destination=ThreadTarget(self.commander.name), + ) + + +@dataclass(frozen=True) +class CommanderNotice(CommanderUpdate): + role = NoticeRole() + + def body(self, previous: CommanderUpdate | None) -> str: + return "\n".join(("Commander observations updated; no new decision.", + *self.changed_lines(previous))) + + +@dataclass(frozen=True) +class CommanderDecisions(CommanderUpdate): + decisions: tuple[Ping, ...] + role = ConversationRole() + + def body(self, previous: CommanderUpdate | None) -> str: + also_current = self.changed_pings(previous, self.decisions) + return "\n".join((*(ping.text for ping in self.decisions), + *(("ALSO CURRENT:", *(ping.text for ping in also_current)) + if also_current else ()), + *self.changed_lines(previous, self.handled))) diff --git a/src/agent_comms/connector.py b/src/agent_comms/connector.py index be446f1813..f060522c9e 100644 --- a/src/agent_comms/connector.py +++ b/src/agent_comms/connector.py @@ -67,14 +67,13 @@ from .commander_heartbeat import CommanderHeartbeat from .coordinator import Coordination from .service_passes import ServicePassStore -from .head_review import HeadReview, PullRequestHead -from .repository_name import RepositoryName -from .todos import DirectiveText, ReviewedHeadVerdict +from .head_review import PullRequestHead +from .todos import DirectiveText, HeadReviewOrder from .tool_result import ToolResult from .task_sources import OwnerRequestAttachment from .tool_caller import ThreadCaller, ToolCaller from .tools import CommsSendTool, SendTool -from .order_tools import CommsVerdictTool, MessagePublished, TodoChanged +from .order_tools import CommsHeadVerdictTool, HeadVerdictTool, MessagePublished from .tool_request import ReadTool, ToolRequest, tool_field LOOPBACK = "127.0.0.1" @@ -94,6 +93,10 @@ def require_directive(self, comms: Comms, body: str) -> None: def origin(self) -> DirectorOrigin: return DirectorOrigin() + def head_review_actor(self, comms: Comms, request: PullRequestHead, + order: HeadReviewOrder): + return order.director_identity(comms.goals, self.identity(comms)) + class ConnectorTool(ToolRequest[DirectorCaller], affix="ConnectorTool"): """Connector-only declarations are absent from the native/CLI tool registry.""" @@ -112,46 +115,9 @@ def attachment(self, comms: Comms, target: NamedTarget) -> OwnerRequestAttachmen @dataclass(frozen=True, kw_only=True) -class ConnectorVerdictTool(ConnectorTool, declared_name=CommsVerdictTool.declared_name): - label = "Review PR head" - description = "Record the connector reviewer's acceptance of the exact PR commit it reviewed." - id: Annotated[str, *tool_field("Problem ID")] - operation: Annotated[ReviewedHeadVerdict, *tool_field("Accepted PR-head verdict")] - - def apply(self, comms: Comms, caller: DirectorCaller) -> TodoChanged: - todo = self.operation.reviewed( - comms.goals, self.id, comms.registry.require(caller.name).incarnation - ) - return TodoChanged(todo, todo.strike_count) - - -@dataclass(frozen=True) -class HeadReviewed(ToolResult): - repo: RepositoryName - pr: int - head: str - review: HeadReview - - -@dataclass(frozen=True, kw_only=True) -class ConnectorHeadReviewTool(ConnectorTool, declared_name="comms_review_head"): - """A pull request with no order still gets the connector reviewer's verdict on its head; - an order's pull request takes comms_verdict, which also records the verdict on the order.""" - - label = "Review PR head without an order" - description = ( - "Post the connector reviewer's verdict (accepted, or sent_back with its reason) as " - "agent-comms/review on the exact head commit of a pull request no order holds." - ) - repo: Annotated[RepositoryName, *tool_field("Repository, owner/name")] - pr: Annotated[int, *tool_field("Pull request number")] - head: Annotated[str, GitCommitText, *tool_field("The exact head commit reviewed")] - review: Annotated[HeadReview, *tool_field("accepted, or sent_back with the reason")] - - def apply(self, comms: Comms, caller: DirectorCaller) -> HeadReviewed: - head = PullRequestHead(self.repo, self.pr, self.head).current() - self.review.post(self.repo, head, caller.name, f"#{self.pr}") - return HeadReviewed(self.repo, self.pr, self.head, self.review) +class ConnectorHeadVerdictTool(HeadVerdictTool, ConnectorTool, + declared_name=CommsHeadVerdictTool.declared_name): + """Director and native callers share the same persisted verdict and publication path.""" CALLER = DirectorCaller("claude-ai") # Owner-typed calls act as this registered bus thread. diff --git a/src/agent_comms/external_transcript.py b/src/agent_comms/external_transcript.py index c104f2537a..d65324df5a 100644 --- a/src/agent_comms/external_transcript.py +++ b/src/agent_comms/external_transcript.py @@ -242,7 +242,7 @@ def append(cls, connection, thread, seq, events) -> None: cls(thread, seq, ordinal, event).insert(connection) @classmethod - def unjudged_query(cls, thread, todo): + def unjudged_query(cls, thread, todo, since): from .native_tools import MutatingPathTool from .tool_declaration import ToolDeclaration from .transcript_events import ToolStartTranscript @@ -250,7 +250,6 @@ def unjudged_query(cls, thread, todo): tools = tuple(tool.declared_name for tool in ToolDeclaration.members_with(MutatingPathTool)) judged = todo.judged_calls placeholders = lambda values: ",".join("?" for _ in values) - since = todo.unjudged_since where = (f'e.thread=? AND e.kind=? AND e.tool IN ({placeholders(tools)}) ' 'AND (e.timestamp>=? OR e.timestamp IS NULL)') parameters = (thread, ToolStartTranscript.declared_name, *tools, since) @@ -414,8 +413,8 @@ def tool_call_count(self, since: float) -> int: FieldCodec.encode(ToolStartTranscript), since), ).fetchone()[0] - def unjudged(self, todo): - where, parameters = ExternalEventRow.unjudged_query(self.transcript.lookup, todo) + def unjudged(self, todo, since): + where, parameters = ExternalEventRow.unjudged_query(self.transcript.lookup, todo, since) with closing(self.connection.execute( 'SELECT timestamp,call_id,json_extract(event,\'$.tool\') FROM ' '(SELECT e.event, e.timestamp, e.call_id, e.seq, e.ordinal, ' @@ -429,8 +428,8 @@ def unjudged(self, todo): yield from tool.edit_calls(identity, GeneratedTimestamp.decode(timestamp), todo.held().owner, todo.id) - def unjudged_count(self, todo) -> int: - where, parameters = ExternalEventRow.unjudged_query(self.transcript.lookup, todo) + def unjudged_count(self, todo, since) -> int: + where, parameters = ExternalEventRow.unjudged_query(self.transcript.lookup, todo, since) return self.connection.execute( f'SELECT count(DISTINCT e.call_id) FROM "{ExternalEventRow.declared_name}" e WHERE {where}', parameters).fetchone()[0] diff --git a/src/agent_comms/field_codec.py b/src/agent_comms/field_codec.py index fe35badfbd..f509a79799 100644 --- a/src/agent_comms/field_codec.py +++ b/src/agent_comms/field_codec.py @@ -1617,6 +1617,16 @@ def digest(cls, value: object) -> str: """SHA-256 of a declared value's sorted compact encoding.""" return hashlib.sha256(cls.compact(cls.encode(value), sort_keys=True).encode()).hexdigest() + @classmethod + def comparison_digest(cls, value: object) -> str: + """Stable identity of declared comparison fields, including nested records. + + This is an observation key, not another wire format or load path. Fields + excluded from dataclass comparison never become notification identity. + """ + compared = compile_decoder(None).comparison(value) + return hashlib.sha256(cls.compact(compared, sort_keys=True).encode()).hexdigest() + @classmethod def encode_document(cls, value: object, declared: object) -> object: """A document whose declaration names its leaf member stores that member untagged.""" @@ -2035,7 +2045,14 @@ def annotation_class(cls, annotation: object) -> type[ValueDecoder[object]]: try: return cls._by_annotation[annotation] except KeyError: - return cls._by_origin[get_origin(annotation) or annotation] + pass + declarations = annotation.__mro__ if isinstance(annotation, type) else (annotation,) + for declaration in declarations: + try: + return cls._by_origin[get_origin(declaration) or declaration] + except KeyError: + continue + raise KeyError(annotation) @classmethod def with_representation( @@ -2069,6 +2086,14 @@ def decode(self, data: object) -> T: @abstractmethod def encode(self, value: T) -> object: ... + def comparison(self, value: T) -> object: + """The nominal value and its declaration-owned comparison content.""" + declaration = type(value) + return (declaration.__module__, declaration.__qualname__, self.compared(value)) + + def compared(self, value: T) -> object: + return self.encode(value) + def decode_external(self, data: object, record: Mapping[str, object]) -> T: for decode in self.external_decoders(data): return self.decode_resolved(cast(T, decode(data, record))) @@ -2195,6 +2220,12 @@ def encode(self, value: T) -> T: return value +class _PathDecoder(PathText, ValueDecoder[Path]): + """The path representation also owns direct and concrete-subclass values.""" + + origins = (Path,) + + class _NullDecoder(_ExactScalarDecoder[None]): __slots__ = () annotations = (type(None),) @@ -2290,6 +2321,10 @@ def argument_options(self, argument: Argument) -> dict[str, object]: def encode(self, value: T) -> object: return None if value is None else self.representation.encode(value) + def compared(self, value: T) -> object: + # Wire spellings (including rename labels) do not own value comparison. + return _structural_decoder(type(value)).compared(value) + def accepts_value(self, value: object) -> bool: return self.structural.accepts_value(value) @@ -2345,6 +2380,13 @@ def encode(cls, value: object) -> object: return [[cls.encode(key), cls.encode(item)] for key, item in value.items()] return compile_decoder(type(value)).encode(value) + def compared(self, value: object) -> object: + if isinstance(value, type): + return value.__module__, value.__qualname__ + if type(value) is dict: + return _PairKeyDictDecoder(dict, self, self).compared(value) + return compile_decoder(type(value)).compared(value) + def _declared_member[Family: DeclaredFamily]( family: type[Family], name: object, @@ -2486,12 +2528,18 @@ def argument_options(self, argument: Argument) -> dict[str, object]: def accepts_value(self, value: object) -> bool: return any(alternative.accepts_value(value) for alternative in self.alternatives) - def encode(self, value: T) -> object: + def selected(self, value: T) -> ValueDecoder[T]: for alternative in self.alternatives: if alternative.accepts_value(value): - return alternative.encode(value) + return alternative raise TypeError(f"Value does not match {self.target}") + def encode(self, value: T) -> object: + return self.selected(value).encode(value) + + def compared(self, value: T) -> object: + return self.selected(value).compared(value) + def decode_bound(self, data: object, context: Mapping[str, object]) -> T: return self.decode_candidates( data, lambda alternative: alternative.decode_bound(data, context), @@ -2576,6 +2624,9 @@ def accepts_value(self, value: object) -> bool: def encode(self, value: T) -> list[object]: return self.encode_items(value) + def compared(self, value: T) -> list[object]: + return [plan.comparison(item) for plan, item in self.plan_items(list(value))] + def decode(self, data: object) -> T: return self.decode_items(data, lambda plan, value: plan.decode_bound(value, {})) @@ -2601,6 +2652,9 @@ class _FrozenSetDecoder[E](_ArrayDecoder[frozenset[E]]): def encode(self, value: frozenset[E]) -> list[object]: return self.encode_items(sorted(value)) + def compared(self, value: frozenset[E]) -> list[object]: + return sorted(super().compared(value), key=lambda item: FieldCodec.compact(item, sort_keys=True)) + class _ListDecoder[E](_ArrayDecoder[list[E]]): origins = (list,) @@ -2710,6 +2764,11 @@ def representation_plan( def accepts_value(self, value: object) -> bool: return isinstance(value, dict) + def compared(self, value: dict[K, V]) -> list[object]: + entries = [(self.key.comparison(key), self.value.comparison(item)) + for key, item in value.items()] + return sorted(entries, key=lambda item: FieldCodec.compact(item[0], sort_keys=True)) + def decode(self, data: object) -> dict[K, V]: return self.decode_entries( data, lambda key: self.key.decode_bound(key, {}), @@ -2801,6 +2860,9 @@ def _rewrite_records( def encode(self, value: Member) -> dict[str, object]: return {**value.record_tag(), **_member_decoder(type(value)).encode(value)} + def compared(self, value: Member) -> object: + return _member_decoder(type(value)).compared(value) + def editor_choices(self): return tuple((member.label, {self.target.family_discriminator: member.declared_name}) for member in self.target.members_with(self.target)) @@ -2907,6 +2969,13 @@ def encode(self, value: object) -> dict[str, object]: ) return result + def compared(self, value: T) -> object: + # Do not compile a full wire plan: excluded fields can be live readers + # that intentionally have no wire representation. + dynamic = compile_decoder(None) + return tuple((item.name, dynamic.comparison(getattr(value, item.name))) + for item in fields(value) if item.compare) + def problems(self, data: dict) -> dict[str, str]: plan = self.plan missing = plan.required_wire_keys.difference(data) diff --git a/src/agent_comms/goal_actions.py b/src/agent_comms/goal_actions.py index 0eebe859cf..4a54c5966f 100644 --- a/src/agent_comms/goal_actions.py +++ b/src/agent_comms/goal_actions.py @@ -23,7 +23,9 @@ UnorderedGoalOrder, ) from agent_comms.input_disposition import InputDispositions -from agent_comms.todos import PendingProblemOrder, SendBackVerdict, Todo, TodoOperation +from agent_comms.todos import ( + PendingProblemOrder, RecordOrderHeadVerdict, SendBackVerdict, Todo, TodoCommand, TodoOperation, +) from .change_report import FileLine, NoSourceReference from .child_process import ProcessIdentity @@ -397,10 +399,10 @@ def require_claims(cls, claims: tuple[ProgressClaim, ...]) -> None: class HeldOrderGoalAction(GoalAction[Goal], RuntimeInvocable): """A Todo operation on a held order and the revision of the holder's Goal, in one write.""" problem: str - operation: TodoOperation + operation: TodoCommand - @abstractmethod - def revised(self, goal: Goal, ctx: GoalActionContext) -> Goal: ... + def revised(self, goal: Goal, ctx: GoalActionContext) -> Goal: + return goal def prepare(self, goals, name, actor): ctx = GoalActionContext(goals, goals.registry.require(name), actor) @@ -420,8 +422,10 @@ def change(self, goal: Goal, ctx: GoalActionContext) -> Goal: class SentBackOrderGoalAction(HeldOrderGoalAction): operation: SendBackVerdict - def revised(self, goal: Goal, ctx: GoalActionContext) -> Goal: - return goal + +@dataclass(frozen=True, kw_only=True) +class HeadReviewedOrderGoalAction(HeldOrderGoalAction): + operation: RecordOrderHeadVerdict @dataclass(frozen=True, kw_only=True) @@ -543,8 +547,19 @@ def continue_goal(self, goal: Goal, ctx: GoalActionContext) -> Goal: ProblemGoalOrder.text, goal.id, revision, ctx.thread, ctx.goals.registry.snapshot())) + def prepare(self, goals: Goals, name: str, actor: type) -> ProblemOrderGoalAction: + self.require_reviewed(goals) + return super().prepare(goals, name, actor) + + def require_reviewed(self, goals: Goals) -> None: + """Read remote review facts before the registry lock or child allocation.""" + from .standing_rules import OrderRefusingStandingRule + + if refused := OrderRefusingStandingRule.refusals(goals, self.commander): + raise RelationViolationError("Review overdue heads before sending new orders:\n" + "\n".join(refused)) + def require_ready(self, ctx: GoalActionContext) -> None: - """Refuse an invalid request before goal reservation or allocating a child.""" + """Recheck local request facts before goal reservation or allocating a child.""" self.require_sendable() commander = ctx.goals.registry.require(self.commander) ctx.goals.todos.owner_request(self.request).require_open(commander.incarnation) diff --git a/src/agent_comms/head_review.py b/src/agent_comms/head_review.py index 185b3eedda..7a3b316091 100644 --- a/src/agent_comms/head_review.py +++ b/src/agent_comms/head_review.py @@ -7,10 +7,11 @@ from abc import abstractmethod from dataclasses import dataclass -from typing import ClassVar +from typing import Annotated, ClassVar from .declared_family import DeclaredFamily from .errors import ToolRefusedError +from .field_codec import FieldCodec, GitCommitText from .repository_name import RepositoryName @@ -20,14 +21,19 @@ class PullRequestHead: repo: RepositoryName pr: int - head: str + head: Annotated[str, GitCommitText] + + def __post_init__(self) -> None: + if FieldCodec.decode(int, self.pr) <= 0: + raise ValueError("A pull request number must be positive.") + GitCommitText.decode(self.head) def current(self) -> str: """The PR's full head commit, refused unless the reviewed head is still it.""" read = self.repo.rest(f"pulls/{self.pr}", "--jq", ".head.sha") if read.returncode: raise ToolRefusedError(read.stderr.strip() or f"#{self.pr} could not be read.") - current = read.stdout.strip() + current = GitCommitText.decode(read.stdout.strip()) if not current.startswith(self.head): raise ToolRefusedError( f"#{self.pr} moved to {current[:9]} since {self.head} was reviewed; review again." @@ -59,26 +65,3 @@ def post(self, repo: RepositoryName, head: str, reviewer: str, subject: str) -> ) if posted.returncode: raise ToolRefusedError(posted.stderr.strip() or f"Review status on {head} failed.") - - -@dataclass(frozen=True) -class AcceptedHeadReview(HeadReview): - state = "success" - - def description(self, reviewer: str, subject: str) -> str: - return f"Accepted by {reviewer} on {subject}" - - -@dataclass(frozen=True) -class SentBackHeadReview(HeadReview): - """The head goes back; the reason names the defect, the PR's review comment the detail.""" - - reason: str - state = "failure" - - def __post_init__(self) -> None: - if len(self.reason.split()) < 4: - raise ValueError("Say what is wrong with the head in 4+ words.") - - def description(self, reviewer: str, subject: str) -> str: - return f"Sent back by {reviewer} on {subject}: {self.reason}" diff --git a/src/agent_comms/order_tools.py b/src/agent_comms/order_tools.py index baf9d1f024..cccd165b8d 100644 --- a/src/agent_comms/order_tools.py +++ b/src/agent_comms/order_tools.py @@ -18,7 +18,10 @@ OrderProgress, PlainText, ProgressRead, - ReviewedHeadVerdict, + HeadReviewOrder, + HeadReviewRecord, + HeadVerdict, + NoHeadOrder, ServeTodo, StrikeReason, Todo, @@ -33,6 +36,7 @@ from .goal_actions import CreateProblemGoalAction, OrderGoalAction from .goals import Goal, OrderCheck from .heartbeat_frame import HeartbeatFrame +from .head_review import PullRequestHead from .order_story import FleetStory, OrderStory from .order_work import OrderWork from .plan_phases import OrderPhase @@ -151,26 +155,31 @@ class CommsVerdictTool(CommsChangeProblemTool): ] -@dataclass(frozen=True, kw_only=True) -class CommsHeadVerdictTool(ThreadToolRequest): - """One reviewed commit judges the holder's whole diff: the commander, or any fork of it, - accepts an order's exact PR head, so every edit made before it is judged in one verdict and - the head carries agent-comms/review for the merge queue.""" +@dataclass(frozen=True) +class HeadReviewed(ToolResult): + record: HeadReviewRecord - label = "Accept an order's PR head" + +@dataclass(frozen=True, kw_only=True) +class HeadVerdictTool: + label = "Review a PR head" description = ( - "Accept the exact pull request head commit an order's holder delivered: one verdict " - "judges every edit made before it and posts agent-comms/review on that commit. Quote " - "4+ words of the order it carries out." + "Record acceptance or rejection of an exact repository/PR/head and post " + "agent-comms/review. An order link is optional; linked reviews quote 4+ " + "words of the order, and rejection strikes its holder in the same write." ) - problem: Annotated[str, *tool_field("Problem ID")] - operation: Annotated[ReviewedHeadVerdict, *tool_field("Accepted PR-head verdict")] + request: Annotated[PullRequestHead, *tool_field("Repository, PR number and reviewed head")] + operation: Annotated[HeadVerdict, *tool_field("Accepted or rejected head verdict")] + order: Annotated[HeadReviewOrder, *tool_field("Optional held-order link")] = NoHeadOrder() + + def apply(self, comms: Comms, caller: ThreadCaller) -> HeadReviewed: + actor = caller.head_review_actor(comms, self.request, self.order) + return HeadReviewed(self.operation.reviewed(comms.goals, self.request, actor, self.order)) - def apply(self, comms: Comms, caller: ThreadCaller) -> TodoChanged: - reviewer = caller.identity(comms) - comms.goals.todos.get(self.problem).require_commander(reviewer) - todo = self.operation.reviewed(comms.goals, self.problem, reviewer) - return TodoChanged(todo, todo.strike_count) + +@dataclass(frozen=True, kw_only=True) +class CommsHeadVerdictTool(HeadVerdictTool, ThreadToolRequest): + """The native/CLI member of the shared authenticated head-review operation.""" @dataclass(frozen=True, kw_only=True) diff --git a/src/agent_comms/plan_phases.py b/src/agent_comms/plan_phases.py index a73bee5a20..dacd14f84e 100644 --- a/src/agent_comms/plan_phases.py +++ b/src/agent_comms/plan_phases.py @@ -174,7 +174,10 @@ class ReviewQueuePhase(CommanderNativePhase): """ after = (RecordPlanePhase,) - evidence = RulesetGates() + evidence = AllEvidence(( + ClassDeclared("HeadReviewRecord", ("src/agent_comms/todos.py",)), + RulesetGates(), + )) COMMANDER = "src/agent_comms/commander/" diff --git a/src/agent_comms/registry_document.py b/src/agent_comms/registry_document.py index c1eae19fc0..09f57e9a69 100644 --- a/src/agent_comms/registry_document.py +++ b/src/agent_comms/registry_document.py @@ -33,6 +33,7 @@ UpdatedRegistration, ) from .repo import Repo, RepositorySelection +from .repository_name import RepositoryName from .registry_provenance import ( DeletedThread, RegisteredRecipient, @@ -495,6 +496,9 @@ class CommanderProject(DeclaredFamily, affix="CommanderProject"): @abstractmethod def commanders(self, snapshot: RegistrySnapshot) -> tuple[Thread, ...]: ... + @abstractmethod + def repository_names(self) -> tuple[RepositoryName, ...]: ... + @abstractmethod def require(self) -> RegisteredCommanderProject: ... @@ -510,6 +514,9 @@ class NoCommanderProject(CommanderProject): def commanders(self, snapshot: RegistrySnapshot) -> tuple[Thread, ...]: return () + def repository_names(self) -> tuple[RepositoryName, ...]: + return () + def require(self) -> RegisteredCommanderProject: raise RelationViolationError("This bus has no registered commander project.") @@ -533,6 +540,9 @@ def commanders(self, snapshot: RegistrySnapshot) -> tuple[Thread, ...]: raise RelationViolationError("The registered commander thread no longer exists.") return selected + def repository_names(self) -> tuple[RepositoryName, ...]: + return tuple(repo.github for repo in self.repos) + def require(self) -> RegisteredCommanderProject: return self diff --git a/src/agent_comms/repo.py b/src/agent_comms/repo.py index 2c5211c6c0..30a9d94aee 100644 --- a/src/agent_comms/repo.py +++ b/src/agent_comms/repo.py @@ -83,6 +83,11 @@ def from_wire(cls, data: object) -> Self: @classmethod def selected(cls, repositories: tuple[Repo, ...], name: str) -> Repo: + if not name: + if len(repositories) != 1: + raise ValueError("Select --repo when more than one repository is registered.") + repository, = repositories + return repository for repository in repositories: if repository.named(name): return repository diff --git a/src/agent_comms/service_passes.py b/src/agent_comms/service_passes.py index 34aa7b0e26..8e10062e98 100644 --- a/src/agent_comms/service_passes.py +++ b/src/agent_comms/service_passes.py @@ -1,13 +1,15 @@ -"""Each Core service watch whose newest pass failed, readable from every process on the bus.""" +"""Original completion times and current failures of Core's service watches.""" from __future__ import annotations import time +from collections.abc import Iterable from dataclasses import dataclass, field, replace from typing import ClassVar -from .locked_store import DocumentChange, LockedStore, RuntimeFile +from .locked_store import DocumentChange, DurableFile, LockedStore from .store_files import DependentStateTier +from .thread_identity import RecordedThreadIncarnation, ThreadIncarnation @dataclass(frozen=True) @@ -21,10 +23,12 @@ class FailingPass: @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.""" @dataclass(frozen=True, slots=True) -class ServicePassStore(RuntimeFile, LockedStore[ServicePasses], DependentStateTier): +class ServicePassStore(DurableFile, LockedStore[ServicePasses], DependentStateTier): """The connector's log reaches nobody; the commander's standing rules read this file.""" filename: ClassVar[str] = "service_passes.json" @@ -47,8 +51,22 @@ def passed(self, watch: str) -> None: self.update( lambda passes: DocumentChange.to(replace( passes, - failing={ - name: failing for name, failing in passes.failing.items() if name != watch - }, + failing={name: failure for name, failure in passes.failing.items() if name != watch}, )) ) + + def heartbeat_passed(self, commanders: Iterable[ThreadIncarnation], + observed: Iterable[ThreadIncarnation]) -> None: + current = frozenset(observed) + completed = {owner.require_recorded(): time.time() for owner in commanders} + + def change(passes: ServicePasses) -> DocumentChange[ServicePasses, ServicePasses]: + retained = {owner: at for owner, at in passes.commander_completed.items() + if owner in current} + updated = retained | completed + return DocumentChange.to( + passes if updated == passes.commander_completed + else replace(passes, commander_completed=updated) + ) + + self.update(change) diff --git a/src/agent_comms/standing_rules.py b/src/agent_comms/standing_rules.py index cd15bcaa7e..854e2e0372 100644 --- a/src/agent_comms/standing_rules.py +++ b/src/agent_comms/standing_rules.py @@ -9,18 +9,21 @@ import logging import time from abc import abstractmethod -from collections import Counter, defaultdict +from collections import Counter from collections.abc import Iterable, Mapping, Sequence -from dataclasses import dataclass, field +from dataclasses import dataclass, field, fields from pathlib import Path from itertools import combinations from typing import TYPE_CHECKING, ClassVar, Self from agent_comms.declared_family import DeclaredFamily +from agent_comms.field_codec import FieldCodec from agent_comms.goals import GoalChangedError +from agent_comms.text_digest import TextDigest from agent_comms.recorded_time import EventTime, RecordedEventTime from agent_comms.repo import Repo from agent_comms.repository_name import RepositoryName +from agent_comms.head_review import PullRequestHead from agent_comms.thread_identity import ThreadIncarnation from agent_comms.work_branches import BranchDifference from agent_comms.service_passes import ServicePassStore @@ -33,7 +36,8 @@ from agent_comms.messages import Message from agent_comms.registry_document import RegistrySnapshot from agent_comms.storage_rules import MemoryUse, StorageRule - from agent_comms.todos import OrderClaim, OwnerRequest, PullRequest, Todo + from agent_comms.todos import OrderClaim, OwnerRequest, ProgressRead, PullRequest, Todo + from agent_comms.goal_management import Goals from agent_comms.ui_model.unit_state import UnitObservation, UnitState @@ -124,37 +128,18 @@ def enforce_all(comms: Comms, commander: str, rejected.add(rule) return frozenset(rejected) - def heartbeat_group(self) -> object: - """Ordinary violations each own one heartbeat line.""" - return self + def comparison_values(self) -> tuple[object, ...]: + """The declaration's equality fields, excluding presentation and live readers.""" + return tuple(getattr(self, item.name) for item in fields(self) if item.compare) - def heartbeat_line(self, current: Sequence[Self]) -> str: - return self.line() + def observation_key(self) -> TextDigest: + """Stable content identity of this occurrence, never a digest of age or wording. - def heartbeat_change(self, previous: Sequence[Self], current: Sequence[Self]) -> tuple[str, ...]: - prefix = "RULE" if current else "NO LONGER CURRENT:" - return (f"{prefix} {self.heartbeat_line(current or previous)}",) - - @staticmethod - def _heartbeat_groups(rules: Iterable[StandingRule]) -> dict[object, list[StandingRule]]: - groups: dict[object, list[StandingRule]] = defaultdict(list) - for rule in rules: - groups[rule.heartbeat_group()].append(rule) - return groups - - @classmethod - def heartbeat_changes(cls, previous: Iterable[StandingRule], - current: Iterable[StandingRule]) -> tuple[str, ...]: - """Render changed groups through their owning rule, without replaying unchanged ones.""" - before, after = cls._heartbeat_groups(previous), cls._heartbeat_groups(current) - lines = [] - for group in dict.fromkeys((*after, *before)): - old, new = before.get(group, []), after.get(group, []) - if frozenset(old) == frozenset(new): - continue - rule = (new or old)[0] - lines.extend(rule.heartbeat_change(old, new)) - return tuple(lines) + Python's hash changes across processes. The codec gives the comparison + values their declared encoding; the class's own name distinguishes siblings + even when they deliberately have no public StandingRule registration. + """ + return TextDigest(FieldCodec.comparison_digest((type(self), self.comparison_values()))) @classmethod def scan_owner(cls) -> type[StandingRule]: @@ -172,6 +157,28 @@ def broken(cls, comms: Comms, commander: str, after: float = 0.0, *, if isinstance(rule, cls) and rule.attention() and rule.broke_after(after)) +class OrderRefusingStandingRule(StandingRule, declared_name=None): + """An overdue review prevents new orders, through the same backend admission path.""" + + @classmethod + @abstractmethod + def scan_reviews(cls, goals: Goals, commander: str, pulls: ProgressRead) -> tuple[Self, ...]: ... + + @classmethod + def scan_observed(cls, comms: Comms, commander: str, + observed: Sequence[UnitState], frame: HeartbeatFrame) -> tuple[Self, ...]: + return cls.scan_reviews(comms.goals, commander, frame.pulls) + + @classmethod + def refusals(cls, goals: Goals, commander: str) -> tuple[str, ...]: + from agent_comms.todos import ProgressRead + + pulls = ProgressRead(goals.todos.held_for_commander(goals.registry.require(commander).incarnation)) + owners = dict.fromkeys(member.scan_owner() for member in cls.members_with(cls)) + return tuple(rule.line() for owner in owners + for rule in owner.scan_reviews(goals, commander, pulls)) + + class IndependentStandingRule(StandingRule, declared_name=None): """A rule whose source read does not consume the unit observation frame.""" @@ -211,9 +218,6 @@ def scan_observed(cls, comms: Comms, commander: str, def line(self) -> str: return f"{self.thread.name}: {'; '.join(self.changes)}" - def heartbeat_change(self, previous: Sequence[Self], current: Sequence[Self]) -> tuple[str, ...]: - return super().heartbeat_change(previous, current) if current else () - @dataclass(frozen=True) class HeldBranch: @@ -545,6 +549,34 @@ def line(self) -> str: "-> commander reviews this head") +@dataclass(frozen=True) +class UnreviewedHead(OrderRefusingStandingRule, DatedStandingRule): + """One ready head, including an orderless PR, waiting too long for a recorded verdict.""" + request: PullRequestHead + due_at: float + seconds: ClassVar[int] = 10 * 60 + + @classmethod + def scan_reviews(cls, goals: Goals, commander: str, pulls: ProgressRead) -> tuple[Self, ...]: + now = time.time() + return tuple( + cls(request, due) + for repo in goals.registry.snapshot().commander_project.repository_names() + for row in pulls.open(repo) + for due in row.review_due(now, cls.seconds) + for request in (PullRequestHead(repo, row.number, row.head),) + if not goals.todos.head_reviews(request) + ) + + def began(self) -> float: + return self.due_at + + def line(self) -> str: + return (f"UNREVIEWED HEAD: {self.request.repo.to_wire()}#{self.request.pr} " + f"head {self.request.head} -> judge ok/back --pr {self.request.pr} " + f"--head {self.request.head}") + + @dataclass(frozen=True) class HuntDue(ReasonedStandingRule, DatedStandingRule): since: float @@ -642,19 +674,6 @@ def line(self) -> str: f" due {self.directive.due_time} -> order its next step or mark the request done" ) - def heartbeat_group(self) -> object: - return type(self), self.request - - def heartbeat_line(self, current: Sequence[Self]) -> str: - oldest = min(current, key=lambda rule: rule.directive.seq).directive - newest = max(current, key=lambda rule: rule.directive.seq).directive - return ( - f"DIRECTIVE UNACTED: request {self.request.value}: {len(current)} owner directives; " - f"oldest {oldest.seq} due {oldest.due_time}; " - f"newest {newest.seq} due {newest.due_time} " - "-> order the next step or mark the request done" - ) - @dataclass(frozen=True) class BridgeDown(ReasonedStandingRule, UndatedStandingRule): diff --git a/src/agent_comms/todos.py b/src/agent_comms/todos.py index 0bc552cfb4..42590dfa8d 100644 --- a/src/agent_comms/todos.py +++ b/src/agent_comms/todos.py @@ -68,7 +68,7 @@ ) from agent_comms.goals import CheckedCommit, OrderCheck, ProblemGoalOrder from agent_comms.repo import GitCommandError -from agent_comms.head_review import AcceptedHeadReview +from agent_comms.head_review import PullRequestHead from agent_comms.lifecycle_capability import LifecycleCapability, Pending from agent_comms.message_reference import MessageReference from agent_comms.messages import InfoKind, Message @@ -592,11 +592,24 @@ class ListedPullRequest(DeclaredFamily, affix="ListedPullRequest"): head: str updated: str base: str + reviewable_times: tuple[str, ...] = field(kw_only=True) draft: bool = field(default=False, kw_only=True) status_contexts: tuple[str, ...] = field(default=(), kw_only=True) + @cached_property + def updated_at(self) -> float: + return _github_time(self.updated) + + @cached_property + def reviewable_at(self) -> float: + return max(map(_github_time, self.reviewable_times)) + + def review_due(self, now: float, seconds: float) -> tuple[float, ...]: + due = self.reviewable_at + seconds + return () if self.draft or now < due else (due,) + def parsed(self, problems: tuple[str, ...], orders: tuple[str, ...]) -> PullRequest: - return UnmergedPullRequest(self.number, self.branch, self.head, _github_time(self.updated), + return UnmergedPullRequest(self.number, self.branch, self.head, self.updated_at, orders, base=self.base, problems=problems, draft=self.draft, status_contexts=self.status_contexts) @@ -615,7 +628,7 @@ class MergedListedPullRequest(ListedPullRequest): merged: str def parsed(self, problems: tuple[str, ...], orders: tuple[str, ...]) -> MergedPullRequest: - return MergedPullRequest(self.number, self.branch, self.head, _github_time(self.updated), + return MergedPullRequest(self.number, self.branch, self.head, self.updated_at, orders, base=self.base, merge=self.merge.oid, merged_at=_github_time(self.merged), problems=problems, draft=self.draft, @@ -626,7 +639,8 @@ class PullRequests: """GitHub facts for recorded links, cached within one board read.""" def __init__(self, todos: Sequence[Todo] = ()) -> None: - self._read: dict[tuple[RepositoryName, type[ListedPullRequest], str], tuple[PullRequest, ...]] = {} + self._read: dict[tuple[RepositoryName, int], ListedPullRequest] = {} + self._open: dict[RepositoryName, tuple[int, ...]] = {} self._todos: dict[RepositoryName, dict[str, Todo]] = {} self.register(todos) @@ -650,41 +664,48 @@ def claimed_heads(self, todo: Todo) -> Iterator[tuple[OrderClaim, PullRequest]]: if request.number == claim.pull_request.number: yield claim, request - def listed(self, repo: RepositoryName, kind: type[ListedPullRequest], - todo: Todo) -> tuple[PullRequest, ...]: - key = (repo, kind, todo.id) - if key not in self._read: - self._read[key] = self.read(repo, kind, todo) - return tuple(request for request in self._read[key] if request.belongs_to(todo)) - - def read(self, repo: RepositoryName, kind: type[ListedPullRequest], - todo: Todo) -> tuple[PullRequest, ...]: - problems = self._todos.setdefault(repo, {}) - problems[todo.id] = todo + def linked(self, repo: RepositoryName) -> dict[int, list[Todo]]: linked: dict[int, list[Todo]] = {} - for candidate in problems.values(): - for number in {link.number for order in candidate.order_history + for candidate in self._todos.get(repo, {}).values(): + for number in {link.number for order in candidate.order.declarations for link in (*order.pull_requests, *(claim.pull_request for claim in order.commander_claims())) if link.repo == repo}: linked.setdefault(number, []).append(candidate) - if not linked: - for candidate in problems.values(): - self._read[repo, kind, candidate.id] = () - return () - requests = tuple((row, row.parsed( + return linked + + def rows(self, repo: RepositoryName, numbers: tuple[int, ...]) -> tuple[ListedPullRequest, ...]: + missing = tuple(number for number in numbers if (repo, number) not in self._read) + if missing: + self._read.update(((repo, row.number), row) for row in self.query(repo, missing)) + return tuple(self._read[repo, number] for number in numbers) + + def open(self, repo: RepositoryName) -> tuple[UnmergedListedPullRequest, ...]: + """Discover orderless open identities, then use the same metadata owner as linked PRs.""" + if repo not in self._open: + result = repo.rest("pulls", "--paginate", "--jq", ".[].number") + if result.returncode: + raise TodoError(result.stderr.strip() or "Open pull requests could not be read.") + self._open[repo] = tuple(FieldCodec.decode(int, FieldCodec.parse(line)) + for line in result.stdout.splitlines()) + numbers = tuple(dict.fromkeys((*self._open[repo], *self.linked(repo)))) + self.rows(repo, numbers) + return tuple(row for number in self._open[repo] + for row in (self._read[repo, number],) + if isinstance(row, UnmergedListedPullRequest)) + + def listed(self, repo: RepositoryName, kind: type[ListedPullRequest], + todo: Todo) -> tuple[PullRequest, ...]: + self.register((todo,)) + linked = self.linked(repo) + requests = tuple(row.parsed( tuple(candidate.id for candidate in linked[row.number]), tuple(dict.fromkeys(order.id for candidate in linked[row.number] - for order in candidate.order_history - if PullRequestLink(repo, row.number) in order.pull_requests)))) - for row in self.query(repo, tuple(linked))) - for selection in dict.fromkeys((kind, *ListedPullRequest.members_with(ListedPullRequest))): - for candidate in problems.values(): - self._read[repo, selection, candidate.id] = tuple(sorted( - (request for row, request in requests - if isinstance(row, selection) and request.belongs_to(candidate)), - key=lambda request: request.updated_at, reverse=True)) - return self._read[repo, kind, todo.id] + for order in candidate.order.declarations + if PullRequestLink(repo, row.number) in order.pull_requests))) + for row in self.rows(repo, tuple(linked)) if isinstance(row, kind)) + return tuple(sorted((request for request in requests if request.belongs_to(todo)), + key=lambda request: request.updated_at, reverse=True)) def query(self, repo: RepositoryName, numbers: tuple[int, ...]) -> tuple[ListedPullRequest, ...]: @@ -692,8 +713,10 @@ def query(self, repo: RepositoryName, selections = [ f'p{number}: pullRequest(number: {number}) {{ number branch: headRefName ' 'head: headRefOid base: baseRefName draft: isDraft closed merge: mergeCommit { oid } ' - 'merged: mergedAt updated: updatedAt ' - 'head_statuses: commits(last: 1) { nodes { commit { status { contexts { context } } } } } }' + 'merged: mergedAt updated: updatedAt created: createdAt ' + 'ready_events: timelineItems(itemTypes: [READY_FOR_REVIEW_EVENT], last: 1) ' + '{ nodes { ... on ReadyForReviewEvent { createdAt } } } ' + 'head_statuses: commits(last: 1) { nodes { commit { committedDate status { contexts { context } } } } } }' for number in numbers] query = (f'query {{ repository(owner: {json.dumps(repo.owner)}, ' f'name: {json.dumps(repo.name)}) {{ {" ".join(selections)} }} }}') @@ -704,8 +727,10 @@ def query(self, repo: RepositoryName, f'del(.merge,.merged) + {{kind: "{ClosedListedPullRequest.declared_name}"}} else ' f'del(.merge,.merged) + {{kind: "{UnmergedListedPullRequest.declared_name}"}} end) else ' f'. + {{kind: "{MergedListedPullRequest.declared_name}"}} end) ' - '| . + {status_contexts: [.head_statuses.nodes[].commit.status.contexts[]?.context]} ' - '| del(.closed,.head_statuses))') + '| . + {status_contexts: [.head_statuses.nodes[].commit.status.contexts[]?.context], ' + 'reviewable_times: [.created, .head_statuses.nodes[].commit.committedDate, ' + '.ready_events.nodes[].createdAt]} ' + '| del(.closed,.created,.head_statuses,.ready_events))') if listed.returncode: raise WorkBranchError( listed.stderr.strip() or f"GitHub order lookup in {repo.to_wire()} failed.") @@ -1894,6 +1919,9 @@ def declarations(self) -> tuple[CheckedProblemOrder, ...]: ) serves: Annotated[type[OrderPhase], WireOmitDefault()] = UnplannedPhase + def holds_head(self, request: PullRequestHead) -> bool: + return PullRequestLink(request.repo, request.pr) in self.pull_requests + @property def phase(self) -> type[OrderPhase]: return self.serves @@ -2731,6 +2759,14 @@ def after_record(self, goals: Goals, changed: Todo) -> None: def validate(self, db: sqlite3.Connection, current: Todo, store: TodoStore) -> None: """Validate related records inside the operation's write transaction.""" + def write(self, db: sqlite3.Connection, current: Todo, store: TodoStore) -> Todo: + """The operation owns its related writes in the Todo transaction.""" + result = self.apply(current) + if result != current: + self.validate(db, current, store) + result.upsert(db) + return result + def apply(self, current: Todo) -> Todo: self.require_current(current) return self.changed(current) @@ -2798,17 +2834,23 @@ def changed(self, current: Todo) -> Todo: @dataclass(frozen=True) -class CitedStrike(TodoOperation, declared_name=None): +class CitedEvidence: evidence: Annotated[tuple[FileLine, ...], WireOmitDefault()] = field( default=(), kw_only=True) - def prepare(self, goals: Goals, current: Todo) -> Self: + def require_evidence(self, current: Todo) -> None: if not self.evidence: raise ValueError("A strike needs a checkable file:line, not a board row or goal text.") order = current.order.checked() root, revision = Path(order.check.directory), order.cited_revision for cited in self.evidence: cited.require_exists(root, revision) + + +@dataclass(frozen=True) +class CitedStrike(CitedEvidence, TodoOperation, declared_name=None): + def prepare(self, goals: Goals, current: Todo) -> Self: + self.require_evidence(current) return self @@ -2828,6 +2870,14 @@ def changed(self, current: Todo) -> Todo: return current return replace(current, strikes=(*current.strikes, self), revision=current.revision + 1) + @classmethod + def send_back(cls, current: Todo, actor: ThreadIncarnation, previous: Assignment, + reason: StrikeReason, at: EventTime, evidence: tuple[FileLine, ...]) -> Todo: + struck = cls(current.revision, actor, previous, reason.text, at, + evidence=evidence).apply(current) + order = current.order.checked().sent_back(reason) + return replace(struck, order=order, prior_orders=current.orders_before(order)) + def story_text(self) -> str: return f"{self.actor.name} struck {self.previous.owner}: {self.reason}" @@ -2900,6 +2950,14 @@ def target(self) -> NamedTarget: name = self.threads[0] if len(self.threads) == 1 else CommsTag.channel() return CommandTarget.named(name) + def publish(self, goals: Goals, actor: ThreadIncarnation, text: str) -> None: + target = self.target + goals.bus.publisher.publish_ordinary( + Message(sender=actor.name, target=target.name, + body=target.addressed(self.threads, text), type=InfoKind), + destination=target, origin=AgentOrigin(), + ) + @dataclass(frozen=True) class OrderVerdict(TodoCommand, StoryEvent, declared_name=None): @@ -2923,10 +2981,6 @@ def judged_calls(self) -> tuple[str, ...]: """The holder's tool calls this verdict judges by identity.""" return () - def judged_through(self, since: float) -> float: - """The instant before which no edit is left unjudged once this verdict is recorded.""" - return since - @abstractmethod def judged(self, current: Todo) -> Todo: ... @@ -2962,17 +3016,20 @@ def require_current(self, current: Todo) -> None: @dataclass(frozen=True) -class QuotedVerdict(OrderVerdict, declared_name=None): +class OrderQuote: quote: str - def story_text(self) -> str: - return f'{super().story_text()}; quoted: "{self.quote}"' - def require_quote(self, text: str) -> None: quoted = " ".join(self.quote.split()) if len(quoted.split()) < 4 or quoted not in " ".join(text.split()): raise ValueError("Quote 4+ words of the order this verdict carries out.") + +@dataclass(frozen=True) +class QuotedVerdict(OrderQuote, OrderVerdict, declared_name=None): + def story_text(self) -> str: + return f'{super().story_text()}; quoted: "{self.quote}"' + def judged(self, current: Todo) -> Todo: self.require_quote(current.order.instructions(current)) return current @@ -2993,29 +3050,220 @@ def require_branches(cls, added: int, no_owner: str | None) -> None: @dataclass(frozen=True) -class ReviewedHeadVerdict(QuotedVerdict): - """The connector reviewer accepts this immutable commit, never an inferred checkout head.""" - head: str +class HeadVerdict(HeadReview, declared_name=None): + """One decision on a PR head, independent of whether an order holds it.""" + + def reviewed(self, goals: Goals, request: PullRequestHead, reviewer: ThreadIncarnation, + order: HeadReviewOrder) -> HeadReviewRecord: + request = replace(request, head=request.current()) + recorded = order.record_review(goals, request, self, reviewer) + self.after_record(goals, recorded) + self.post(request.repo, request.head, reviewer.name, f"#{request.pr}") + return recorded - def __post_init__(self) -> None: - super().__post_init__() - GitCommitText.decode(self.head) + def after_record(self, goals: Goals, record: HeadReviewRecord) -> None: + """A rejection wakes its linked holder after the write; acceptance needs no directive.""" - @property - def line(self) -> str: - return f"Reviewed head {self.head}" - - def judged_through(self, since: float) -> float: - """Accepting a head judges the whole diff, so every edit made before it is judged.""" - return self.at.latest_with(since) - - def reviewed(self, goals: Goals, problem: str, reviewer: ThreadIncarnation) -> Todo: - if self.actor != reviewer: - raise TodoError("The verdict actor must be the connector reviewer recording it.") - current = goals.todos.get(problem) - recorded = self.prepare(goals, current).record(goals, problem) - AcceptedHeadReview().post(recorded.repo, self.head, reviewer.name, problem) - return recorded + def prepare_order(self, current: Todo, link: LinkedHeadReview) -> None: + """Acceptance needs the quoted order; rejection also needs source evidence.""" + + def judge_order(self, current: Todo, record: OrderHeadReviewRecord) -> Todo: + return current + + +@dataclass(frozen=True) +class ReviewedHeadVerdict(HeadVerdict): + state = "success" + + def description(self, reviewer: str, subject: str) -> str: + return f"Accepted by {reviewer} on {subject}" + + +@dataclass(frozen=True) +class RejectedHeadVerdict(HeadVerdict): + reason: StrikeReason + state = "failure" + + def description(self, reviewer: str, subject: str) -> str: + return f"Sent back by {reviewer} on {subject}: {self.reason.text}" + + def after_record(self, goals: Goals, record: HeadReviewRecord) -> None: + record.order.rejected(goals, record) + + def prepare_order(self, current: Todo, link: LinkedHeadReview) -> None: + link.require_evidence(current) + + def judge_order(self, current: Todo, record: OrderHeadReviewRecord) -> Todo: + return StrikeTodo.send_back(current, record.actor, record.order.previous, + self.reason, record.at, record.order.evidence) + + +class HeadReviewRelation(DeclaredFamily): + """Order relationships share one declaration family and compose their capabilities.""" + + +class HeadOrderLink(HeadReviewRelation, declared_name=None): + """The historical order relationship of a recorded head decision.""" + + @abstractmethod + def rejected(self, goals: Goals, record: HeadReviewRecord) -> None: ... + + +class HeadReviewOrder(HeadReviewRelation, declared_name=None): + """The input that authorizes a head review's optional order change.""" + + @classmethod + def for_request(cls, todos: Sequence[Todo], request: PullRequestHead, quote: str, + evidence: tuple[FileLine, ...], problem: str = "") -> HeadReviewOrder: + linked = tuple(todo for todo in todos if todo.repo == request.repo + and (not problem or todo.id == problem) + and todo.order.checked().holds_head(request)) + if not linked: + if problem: + raise TodoError(f"{problem} does not hold this pull request.") + return NoHeadOrder() + if len(linked) != 1: + raise TodoError("More than one order holds this pull request; select --problem.") + todo, = linked + return OrderHeadReview(todo.revision, LinkedHeadReview.of(todo, quote, evidence)) + + @abstractmethod + def require_reviewer(self, goals: Goals, request: PullRequestHead, + actor: ThreadIncarnation) -> None: ... + + def director_identity(self, goals: Goals, actor: ThreadIncarnation) -> ThreadIncarnation: + return actor + + @abstractmethod + def record_review(self, goals: Goals, request: PullRequestHead, verdict: HeadVerdict, + actor: ThreadIncarnation) -> HeadReviewRecord: ... + + +@dataclass(frozen=True) +class NoHeadOrder(HeadReviewOrder, HeadOrderLink): + def rejected(self, goals: Goals, record: HeadReviewRecord) -> None: + """There is no holder to strike or wake.""" + + def require_reviewer(self, goals: Goals, request: PullRequestHead, + actor: ThreadIncarnation) -> None: + project = goals.registry.snapshot().commander_project.require() + if (actor.commanding_identity != project.commander or + request.repo not in {repo.github for repo in project.repos}): + raise TodoError("Only this repository's commander or the director may review its heads.") + + def record_review(self, goals: Goals, request: PullRequestHead, verdict: HeadVerdict, + actor: ThreadIncarnation) -> HeadReviewRecord: + record = HeadReviewRecord(request, actor, RecordedEventTime(time.time()), verdict, self) + goals.todos.record_head(record) + return record + + +@dataclass(frozen=True) +class ProblemHeadReview(OrderQuote, CitedEvidence, HeadOrderLink): + """A problem association without an observed order ID remains an honest partial fact.""" + problem: str + previous: Assignment + + def rejected(self, goals: Goals, record: HeadReviewRecord) -> None: + audience = VerdictAudience.of(self.previous.owner, goals.todos.get(self.problem)) + audience.publish(goals, record.actor, + f"{self.problem}: {record.verdict.description(record.actor.name, f'#{record.request.pr}')}") + + +@dataclass(frozen=True) +class LinkedHeadReview(ProblemHeadReview): + order_id: str + + @classmethod + def of(cls, todo: Todo, quote: str, evidence: tuple[FileLine, ...] = ()) -> Self: + return cls(quote=quote, problem=todo.id, order_id=todo.order.checked().id, + previous=todo.held(), evidence=evidence) + + def require_current(self, current: Todo, actor: ThreadIncarnation) -> None: + current.require_commander(actor) + current.require_assignment(self.previous) + if current.id != self.problem or current.order.checked().id != self.order_id: + raise TodoConflict(current) + self.require_quote(current.order.instructions(current)) + + +@dataclass(frozen=True) +class OrderHeadReview(HeadReviewOrder): + expected_revision: int + link: LinkedHeadReview + + def require_reviewer(self, goals: Goals, request: PullRequestHead, + actor: ThreadIncarnation) -> None: + self.link.require_current(goals.todos.get(self.link.problem), actor) + + def director_identity(self, goals: Goals, actor: ThreadIncarnation) -> ThreadIncarnation: + return actor.require_recorded().acting_for(goals.todos.get(self.link.problem).creator) + + def record_review(self, goals: Goals, request: PullRequestHead, verdict: HeadVerdict, + actor: ThreadIncarnation) -> OrderHeadReviewRecord: + from .goal_actions import GoalPrecondition, HeadReviewedOrderGoalAction + + record = OrderHeadReviewRecord(request, actor, RecordedEventTime(time.time()), verdict, self.link) + operation = RecordOrderHeadVerdict(self.expected_revision, record) + goals.update_goal(self.link.previous.owner, HeadReviewedOrderGoalAction( + expect=GoalPrecondition(goal_id=self.link.previous.generation), + problem=self.link.problem, operation=operation, + )) + return record + + +@dataclass(frozen=True) +class HeadReviewRecord: + request: PullRequestHead + actor: ThreadIncarnation + at: EventTime + verdict: HeadVerdict + order: HeadOrderLink + + +@dataclass(frozen=True) +class OrderHeadReviewRecord(HeadReviewRecord): + order: LinkedHeadReview + + +@dataclass(frozen=True) +class StoredHeadReview(TodoStoreRows, TypedTable): + seq: Annotated[int, Column(primary_key=True, auto_increment=True)] + record: HeadReviewRecord + + request: ClassVar[Generated] = Generated( + PullRequestHead, Column(generated=f"json_extract({{record}}, '$.{FieldCodec.record_key(HeadReviewRecord, 'request')}')"), + ) + problem: ClassVar[Generated] = Generated( + str, Column(generated=f"json_extract({{record}}, '$.{FieldCodec.record_key(HeadReviewRecord, 'order')}.{FieldCodec.record_key(LinkedHeadReview, 'problem')}')", nullable=True), + ) + order_id: ClassVar[Generated] = Generated( + str, Column(generated=f"json_extract({{record}}, '$.{FieldCodec.record_key(HeadReviewRecord, 'order')}.{FieldCodec.record_key(LinkedHeadReview, 'order_id')}')", nullable=True), + ) + indexes = (Index(("request", "seq")), Index(("problem", "order_id", "seq"))) + + +@dataclass(frozen=True) +class RecordOrderHeadVerdict(TodoCommand): + """The order change and its head record share the existing Todo write transaction.""" + review: OrderHeadReviewRecord + + def prepare(self, goals: Goals, current: Todo) -> Self: + self.review.order.require_current(current, self.review.actor) + if not current.order.checked().holds_head(self.review.request): + raise TodoError("The reviewed pull request is not linked to this order.") + self.review.verdict.prepare_order(current, self.review.order) + return self + + def changed(self, current: Todo) -> Todo: + self.review.order.require_current(current, self.review.actor) + judged = self.review.verdict.judge_order(current, self.review) + return replace(judged, revision=current.revision + 1) + + def write(self, db: sqlite3.Connection, current: Todo, store: TodoStore) -> Todo: + changed = super().write(db, current, store) + StoredHeadReview.insert_values(db, record=self.review) + return changed @dataclass(frozen=True) @@ -3122,20 +3370,13 @@ def story_text(self) -> str: return f"{super().story_text()} (strike recorded)" def judged(self, current: Todo) -> Todo: - (strike,) = self.implied_strikes() - struck = strike.apply(current) - order = current.order.checked().sent_back(self.reason) - return replace(struck, order=order, prior_orders=current.orders_before(order)) + return StrikeTodo.send_back(current, self.actor, self.previous, self.reason, + self.at, self.evidence) def record(self, goals: Goals, problem: str) -> Todo: changed = goals.send_back(problem, self) - audience = self.audience.of(self.previous.owner, changed) - target = audience.target - goals.bus.publisher.publish_ordinary( - Message(sender=self.actor.name, target=target.name, - body=target.addressed(audience.threads, f"{changed.id}: {self.story_text()}"), - type=InfoKind), - destination=target, origin=AgentOrigin(), + self.audience.of(self.previous.owner, changed).publish( + goals, self.actor, f"{changed.id}: {self.story_text()}", ) return changed @@ -3486,14 +3727,6 @@ def rows(cls, db: sqlite3.Connection, where: str, parameters: tuple) -> tuple[To def judged_calls(self) -> tuple[str, ...]: return tuple(call for verdict in self.verdicts for call in verdict.judged_calls()) - @property - def unjudged_since(self) -> float: - """Edits before the order, or before a later verdict judging a whole head, are judged.""" - since = self.order.checked().at.recorded_at("An order") - for verdict in self.verdicts: - since = verdict.judged_through(since) - return since - @property def order_history(self) -> tuple[CheckedProblemOrder, ...]: return (*self.prior_orders, *self.order.declarations) @@ -4114,14 +4347,31 @@ def create( ProblemAllocation.delete(db, number=number.number) return current + def record_head(self, record: HeadReviewRecord) -> None: + with self.writing() as db: + StoredHeadReview.insert_values(db, record=record) + + def head_reviews(self, request: PullRequestHead) -> tuple[HeadReviewRecord, ...]: + """An indexed exact-head lookup, newest committed verdict first.""" + with self.snapshot() as db: + return tuple(row.record for row in StoredHeadReview.select( + db, request=request, order_by=("seq",), descending=True, limit=1, + )) + + def unjudged_since(self, todo: Todo) -> float: + order = todo.order.checked() + with self.snapshot() as db: + reviews = StoredHeadReview.select( + db, problem=todo.id, order_id=order.id, + order_by=("seq",), descending=True, limit=1, + ) + began = order.at.recorded_at("An order") + return max((began, *(row.record.at.latest_with(began) for row in reviews))) + def apply(self, todo_id: str, command: TodoCommand) -> Todo: with self.writing() as db: current = self._current(db, todo_id) - result = command.apply(current) - if result == current: - return current - command.validate(db, current, self) - result.upsert(db) + command.write(db, current, self) return self._current(db, todo_id) def correct(self, unit: ThreadIncarnation, message_id: str, text: str) -> tuple[Todo, ...]: diff --git a/src/agent_comms/tool_caller.py b/src/agent_comms/tool_caller.py index 66fc53918a..f3bfc68c9a 100644 --- a/src/agent_comms/tool_caller.py +++ b/src/agent_comms/tool_caller.py @@ -18,6 +18,8 @@ if TYPE_CHECKING: from .comms import Comms from .registration import Registration + from .head_review import PullRequestHead + from .todos import HeadReviewOrder from .threads import Thread from .tool_request import ToolRequest @@ -125,6 +127,12 @@ def commander(self, comms: Comms) -> Thread: def identity(self, comms: Comms) -> ThreadIncarnation: return comms.registry.require(self.name).incarnation + def head_review_actor(self, comms: Comms, request: PullRequestHead, + order: HeadReviewOrder) -> ThreadIncarnation: + actor = self.identity(comms) + order.require_reviewer(comms.goals, request, actor) + return actor + def invoke(self, request: ToolRequest[ThreadCaller], comms: Comms) -> ToolResult: return request.apply(comms, self) diff --git a/src/agent_comms/turn_digests.py b/src/agent_comms/turn_digests.py index 2bbc49e2e9..13fbb9a738 100644 --- a/src/agent_comms/turn_digests.py +++ b/src/agent_comms/turn_digests.py @@ -184,6 +184,7 @@ def from_turns(self, turns: tuple[TurnDigest, ...]) -> ThreadTurns: @dataclass(frozen=True) class OrderRead[Result](TranscriptRead[Result]): + bound: TurnsSince todo: Todo @classmethod @@ -198,19 +199,22 @@ def for_commander(cls, comms: Comms, commander: str) -> Iterator[Self]: def for_orders( cls, root: Path, snapshot: RegistrySnapshot, todos: Iterable[Todo] ) -> Iterator[Self]: - """Read these orders against their captured roster, without reopening the registry.""" + """Read these orders against their captured roster and canonical review cut.""" + from .todos import TodoStore + + store = TodoStore(root) for todo in todos: assignment = todo.held() holder = snapshot.require(assignment.owner) if holder.incarnation != assignment.owner_identity: raise ToolRefusedError(f"Order {todo.id} holder incarnation no longer exists") yield cls( - root, holder, TurnsSince(todo.order.checked().at.recorded_at("An order")), todo + root, holder, TurnsSince(store.unjudged_since(todo)), todo ) def edits_from_turns(self, turns: tuple[TurnDigest, ...]) -> tuple[EditCall, ...]: judged = frozenset(self.todo.judged_calls) - since = self.todo.unjudged_since + since = self.bound.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) @@ -225,7 +229,7 @@ def from_turns(self, turns: tuple[TurnDigest, ...]) -> tuple[EditCall, ...]: return self.edits_from_turns(turns) def from_external(self, source: StoreRead) -> tuple[EditCall, ...]: - return tuple(source.unjudged(self.todo)) + return tuple(source.unjudged(self.todo, self.bound.at)) class OrderEditCountRead(OrderRead[int]): @@ -239,15 +243,13 @@ def from_turns(self, turns: tuple[TurnDigest, ...]) -> int: return len(self.edits_from_turns(turns)) def from_external(self, source: StoreRead) -> int: - return source.unjudged_count(self.todo) + return source.unjudged_count(self.todo, self.bound.at) @dataclass(frozen=True) class OrderToolCallCountRead(OrderRead[int]): """Original tool starts since a captured head/order cut, without loading external bodies.""" - bound: TurnsSince - def from_turns(self, turns: tuple[TurnDigest, ...]) -> int: return len(OrderWork.read( self.todo.id, self.thread.incarnation, self.bound.at, turns, diff --git a/src/agent_comms/work_dispatch.py b/src/agent_comms/work_dispatch.py index 6954ecac50..5426608eca 100644 --- a/src/agent_comms/work_dispatch.py +++ b/src/agent_comms/work_dispatch.py @@ -68,6 +68,7 @@ def start(self, comms: Comms) -> Thread: model=defaults.model, thinking_level=defaults.thinking_level, ) + self.action.require_reviewed(comms.goals) self.action.require_ready(GoalActionContext(comms.goals, candidate)) comms.registry.require_declaration_available(candidate, StoppedThreadStatus()) CheckoutStorageEffect(checkout, Path(self.action.order.check.directory)).check( diff --git a/tests/conftest.py b/tests/conftest.py index de4b3af453..52fab6f0ae 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -49,7 +49,7 @@ from source_revision_fixture import SourceRepository from agent_comms.threads import Thread -pytest_plugins = (goal_owner_fixture.__name__,) +pytest_plugins = (goal_owner_fixture.__name__, "github_fixture") class TranscriptRecords: @@ -339,7 +339,7 @@ def root(tmp_path: Path) -> Path: @pytest.fixture -def configured_repository(durable_home: Path) -> Callable[[Comms, *tuple[Repo, ...]], Commander]: +def configured_repository(durable_home: Path, github_cli) -> Callable[[Comms, *tuple[Repo, ...]], Commander]: """Declare the temporary commander on its bus and use its real plan-bearing checkout.""" from agent_comms.commander.order import Commander from agent_comms.repo import Repo @@ -347,6 +347,7 @@ def configured_repository(durable_home: Path) -> Callable[[Comms, *tuple[Repo, . repository = Repo.of(Path(__file__).resolve().parents[1]) def configure(bus: Comms, *check_repositories: Repo) -> Commander: + github_cli.configure(*(repo.github for repo in (*check_repositories, repository))) return Commander.on_root( 'lead', (*check_repositories, repository), bus.root, ) diff --git a/tests/github_cli_stub.py b/tests/github_cli_stub.py index 8dde8e38f0..cd070725ee 100755 --- a/tests/github_cli_stub.py +++ b/tests/github_cli_stub.py @@ -9,27 +9,58 @@ def main(): directory = Path(sys.argv[0]).parent - data = json.loads((directory / "pull_requests.json").read_text()) + repositories = json.loads((directory / "pull_requests.json").read_text()) arguments = sys.argv[1:] with (directory / "calls.jsonl").open("a") as calls: calls.write(json.dumps(arguments) + "\n") if arguments[:2] == ["pr", "view"]: - assert arguments[arguments.index("-R") + 1] == data["repo"] + data = repositories[arguments[arguments.index("-R") + 1]] request = data["requests"][arguments[2]] field = arguments[arguments.index("--json") + 1] assert arguments[arguments.index("--jq") + 1] == "." + field print({"number": request["listed"]["number"], "body": request["body"]}[field]) + elif len(arguments) > 1 and arguments[0] == "api" and arguments[1].startswith("repos/"): + _, owner, name, *parts = arguments[1].split("/") + data = repositories[f"{owner}/{name}"] + if parts == ["pulls"]: + for entry in data["requests"].values(): + if entry["listed"]["kind"] == data["open_kind"]: + print(entry["listed"]["number"]) + elif parts[:1] == ["pulls"] and len(parts) == 2: + print(data["requests"][parts[1]]["listed"]["head"]) + elif parts[:1] == ["statuses"] and len(parts) == 2: + assert arguments[arguments.index("--method") + 1] == "POST" + print("{}") + else: + raise AssertionError(f"Unexpected repository endpoint: {parts}") elif arguments[:2] == ["api", "graphql"]: query = arguments[arguments.index("-f") + 1] assert query.startswith("query=") - owner, name = data["repo"].split("/") - assert f'repository(owner: "{owner}", name: "{name}")' in query + variables = dict(arguments[index + 1].split("=", 1) + for index, value in enumerate(arguments) if value == "-F") + inline = re.search(r'repository\(owner: "([^"]+)", name: "([^"]+)"\)', query) + repository = "/".join(inline.groups()) if inline else f"{variables['owner']}/{variables['name']}" + data = repositories[repository] selections = re.findall(r"p(\d+): pullRequest\(number: (\d+)\)", query) - assert selections, "Only explicit recorded pull-request selections are supported" - assert all(alias == number for alias, number in selections) - # gh --jq returns the typed projection, not the raw GraphQL envelope. - print(json.dumps({f"p{number}": data["requests"][number]["listed"] - for _, number in selections})) + if selections: + assert all(alias == number for alias, number in selections) + # gh --jq returns the typed projection, not the raw GraphQL envelope. + print(json.dumps({f"p{number}": data["requests"][number]["listed"] + for _, number in selections})) + elif "pullRequests(" in query: + print(json.dumps({"data": {"repository": {"pullRequests": { + "nodes": [{"number": entry["listed"]["number"], + "headRefName": entry["listed"]["branch"], "isCrossRepository": False} + for entry in data["requests"].values() + if entry["listed"]["kind"] == data["open_kind"]], + "pageInfo": {"hasNextPage": False, "endCursor": None}, + }}}})) + elif "issues(" in query: + print(json.dumps({"data": {"repository": {"issues": { + "nodes": [], "pageInfo": {"hasNextPage": False, "endCursor": None}, + }}}})) + else: + raise AssertionError(f"Unexpected GraphQL query: {query}") else: raise AssertionError(f"Unexpected gh invocation: {arguments!r}") diff --git a/tests/github_fixture.py b/tests/github_fixture.py new file mode 100644 index 0000000000..42d357befa --- /dev/null +++ b/tests/github_fixture.py @@ -0,0 +1,48 @@ +"""A real gh subprocess with controlled foreign responses and retained call records.""" + +import json +import os +from dataclasses import dataclass +from pathlib import Path + +import pytest + +from agent_comms.field_codec import FieldCodec +from agent_comms.repository_name import RepositoryName +from agent_comms.todos import ListedPullRequest, UnmergedListedPullRequest + + +@dataclass(frozen=True) +class GitHubFixture: + directory: Path + + @property + def path(self) -> Path: + return self.directory / "pull_requests.json" + + def configure(self, *repositories: RepositoryName): + data = json.loads(self.path.read_text()) if self.path.exists() else {} + for repository in repositories: + data.setdefault(repository.to_wire(), { + "open_kind": UnmergedListedPullRequest.declared_name, "requests": {}, + }) + self.path.write_text(json.dumps(data)) + return self + + def put(self, repository: RepositoryName, row: ListedPullRequest, body: str = "") -> None: + data = json.loads(self.path.read_text()) + data[repository.to_wire()]["requests"][str(row.number)] = {"body": body, "listed": FieldCodec.encode(row)} + self.path.write_text(json.dumps(data)) + + def calls(self) -> tuple[tuple[str, ...], ...]: + path = self.directory / "calls.jsonl" + return tuple(tuple(json.loads(line)) for line in path.read_text().splitlines()) if path.exists() else () + + +@pytest.fixture +def github_cli(tmp_path, monkeypatch): + directory = tmp_path / "bin" + directory.mkdir() + (directory / "gh").symlink_to(Path(__file__).with_name("github_cli_stub.py")) + monkeypatch.setenv("PATH", os.pathsep.join((str(directory), os.environ["PATH"]))) + return GitHubFixture(directory) diff --git a/tests/test_commander_notifications.py b/tests/test_commander_notifications.py new file mode 100644 index 0000000000..a28d4c02cd --- /dev/null +++ b/tests/test_commander_notifications.py @@ -0,0 +1,116 @@ +"""Committed notifications survive a new watcher and use the real publisher/readers.""" + +import os +from dataclasses import replace + +import pytest + +from agent_comms.child_process import ProcessIdentity +from agent_comms.commander_heartbeat import CommanderHeartbeat, CommanderObservation +from agent_comms.commander_pings import CommanderDecisions, CommanderNotice, CommanderUpdate, RulePing +from agent_comms.errors import UnregisteredThreadError +from agent_comms.launch_root import LaunchRoot +from agent_comms.message_page import MessagePageRequest +from agent_comms.read_basis import DMDisplayScope +from agent_comms.service_passes import ServicePassStore +from agent_comms.standing_rules import ServicePassFailing +from agent_comms.ui_model.unit_state import UnitState +from agent_comms.wake_policy import PassiveWake +from test_holder_standing_rules import state_cards + +pytest_plugins = ("test_holder_standing_rules",) + + +@pytest.fixture +def notification_bus(held_bus, installed_runtime): + lead = held_bus.registry.require("lead") + held_bus.registry.register(replace(lead, process_identity=ProcessIdentity.capture(os.getpid()))) + CommanderHeartbeat.register(held_bus.registry) + with LaunchRoot.from_path(held_bus.root).selected(): + yield held_bus + + +def notifications(comms): + page = MessagePageRequest(DMDisplayScope.capture( + CommanderHeartbeat.sender, "lead", comms.registry.snapshot(), + )).read_opened(comms.bus.bus) + return tuple(message for message in page.messages if isinstance(message.task, CommanderUpdate)) + + +def beat(comms, rules=(), *, states=()): + frame = CommanderObservation(comms.registry.require("lead"), (), rules, states) + # Deliberately construct a fresh watcher on every pass, never reuse a cache. + CommanderHeartbeat().pass_body(comms, (frame,)) + + +def test_quiet_completion_is_per_commander_even_when_a_later_recipient_fails(notification_bus): + comms = notification_bus + lead = comms.registry.require("lead") + missing = replace(comms.registry.require("worker"), name="unregistered-recipient") + frames = tuple(CommanderObservation(thread, (), (), ()) for thread in (lead, missing)) + with pytest.raises(UnregisteredThreadError): + CommanderHeartbeat().pass_body(comms, frames) + completed = ServicePassStore.at(comms.root).read().commander_completed + assert set(completed) == {lead.incarnation} + assert missing.incarnation not in completed + # A quiet pass still completes after its original notice was committed. + beat(comms) + assert len(notifications(comms)) == 1 + CommanderHeartbeat().pass_body(comms, ()) + assert not ServicePassStore.at(comms.root).read().commander_completed + + +def test_recreated_watcher_reads_original_committed_unit_observation(notification_bus): + comms = notification_bus + states = state_cards(comms, UnitState) + beat(comms, states=states) + first, = notifications(comms) + assert all(state.line() in first.body for state in states) + assert CommanderUpdate.latest(comms, comms.registry.require("lead").incarnation) == first.task + beat(comms, states=state_cards(comms, UnitState)) + unchanged, = notifications(comms) + assert unchanged.reference == first.reference + + +def test_rule_packets_lead_with_five_and_publish_all_current_occurrences(notification_bus): + comms = notification_bus + rules = tuple(ServicePassFailing(f"watch-{index}", 1.0, "failed source read") + for index in range(7)) + beat(comms, rules) + first, = notifications(comms) + assert isinstance(first.task, CommanderDecisions) + assert len(first.task.decisions) == 5 + assert {RulePing.capture(rule) for rule in rules} == set(first.task.handled) + beat(comms, rules) + assert len(notifications(comms)) == 1 + # Changed diagnostic wording is not another failed occurrence. + beat(comms, tuple(replace(rule, error="new diagnostic detail") for rule in rules)) + assert len(notifications(comms)) == 1 + + +def test_resolution_is_passive_and_same_rule_can_recur(notification_bus): + comms = notification_bus + passes = ServicePassStore.at(comms.root) + passes.failed("dependent-watch", RuntimeError("unreadable source")) + rules = ServicePassFailing.violations(comms, "lead") + beat(comms, rules) + first, = notifications(comms) + passes.passed("dependent-watch") + beat(comms, ServicePassFailing.violations(comms, "lead")) + _, quiet = notifications(comms) + assert isinstance(quiet.task, CommanderNotice) + assert all(isinstance(decision.wake_mode, PassiveWake) for decision in quiet.decisions) + # Use the same source occurrence to prove it was retired, not merely hidden. + beat(comms, rules) + _, _, repeated = notifications(comms) + assert repeated.task.decisions == first.task.decisions + + +def test_latest_observation_follows_commander_and_service_rename_aliases(notification_bus): + comms = notification_bus + beat(comms) + first, = notifications(comms) + comms.threads._rename_thread("lead", "renamed-lead") + comms.threads._rename_thread(CommanderHeartbeat.sender, "renamed-heartbeat") + previous = CommanderUpdate.latest(comms, comms.registry.require("renamed-lead").incarnation) + assert previous == first.task diff --git a/tests/test_delivery_merge.py b/tests/test_delivery_merge.py index 21617545d8..3e857e6646 100644 --- a/tests/test_delivery_merge.py +++ b/tests/test_delivery_merge.py @@ -21,6 +21,7 @@ def test_delivery_integration_members_keep_their_data_through_the_only_codec(): def test_named_forge_row_preserves_the_actual_nonmain_merge_target(): original = MergedListedPullRequest(42, 'fix/order', 'a' * 40, updated='2026-10-19T12:00:01Z', base='release', + reviewable_times=('2026-10-19T12:00:00Z',), merge=PullRequestCommit('b' * 40), merged='2026-10-19T12:00:00Z') row = FieldCodec.decode(ListedPullRequest, FieldCodec.encode(original)) request = row.parsed((), ('proof-order',)) @@ -34,7 +35,8 @@ def test_named_forge_row_preserves_the_actual_nonmain_merge_target(): def test_unmerged_query_row_has_no_merge_fields(): original = UnmergedListedPullRequest(43, 'fix/order', 'a' * 40, - updated='2026-10-19T12:00:01Z', base='main') + updated='2026-10-19T12:00:01Z', base='main', + reviewable_times=('2026-10-19T12:00:00Z',)) row = FieldCodec.decode(ListedPullRequest, FieldCodec.encode(original)) request = row.parsed((), ('proof-order',)) assert isinstance(request, UnmergedPullRequest) diff --git a/tests/test_field_codec_comparison.py b/tests/test_field_codec_comparison.py new file mode 100644 index 0000000000..a1fcc07595 --- /dev/null +++ b/tests/test_field_codec_comparison.py @@ -0,0 +1,49 @@ +"""Observation identity follows comparison declarations, not presentation fields.""" + +from dataclasses import dataclass, field +from enum import IntEnum +from pathlib import Path + +import pytest + +from agent_comms.field_codec import FieldCodec +from agent_comms.thread_identity import RecordedThreadIncarnation + + +@dataclass(frozen=True) +class Observation: + owner: RecordedThreadIncarnation + revision: int + reader: object = field(compare=False, repr=False) + + +@pytest.mark.parametrize("wrap", ( + lambda value: (value,), + lambda value: [value], + lambda value: frozenset((value,)), + lambda value: {"observation": value}, +)) +def test_comparison_recurses_through_container_and_record_declarations(wrap): + original = Observation(RecordedThreadIncarnation("before", 123.0), 1, object()) + renamed = Observation(RecordedThreadIncarnation("after", 123.0), 1, object()) + changed = Observation(RecordedThreadIncarnation("after", 123.0), 2, object()) + assert original == renamed and original != changed + before, after, revised = (FieldCodec.comparison_digest(wrap(value)) + for value in (original, renamed, changed)) + assert before == after and before != revised + + +class Position(IntEnum): + FIRST = 1 + + +@pytest.mark.parametrize("value", (Position.FIRST, 1, True, "value", Path("nested/path"))) +def test_inherited_origin_does_not_replace_exact_scalar_or_enum_declarations(value): + encoded = FieldCodec.encode(value, type(value)) + assert FieldCodec.decode(type(value), encoded) == value + + +def test_unordered_comparison_is_independent_of_insertion_order(): + left = {"first": frozenset(("b", "a")), "second": 2} + right = {"second": 2, "first": frozenset(("a", "b"))} + assert FieldCodec.comparison_digest(left) == FieldCodec.comparison_digest(right) diff --git a/tests/test_fleet_replay.py b/tests/test_fleet_replay.py index 3bac2abd37..af4f922082 100644 --- a/tests/test_fleet_replay.py +++ b/tests/test_fleet_replay.py @@ -1,7 +1,7 @@ """Replay a small fleet written through Core's own writers: each covered incident is raised once. Git is real (a temporary repository with an origin remote and one worktree per unit); only the -forge answer comes from the fleet, through ProgressRead's own query method. +forge answers come from the external gh fixture, through the unmodified reader. """ import os @@ -10,6 +10,7 @@ from abc import ABC, abstractmethod from collections import Counter from dataclasses import dataclass, field, replace +from datetime import UTC, datetime from pathlib import Path from typing import ClassVar @@ -39,11 +40,9 @@ from agent_comms.threads import Thread from agent_comms.todos import ( ClaimOrder, - ListedPullRequest, OrderClaim, OrderMessage, PendingProblemOrder, - ProgressRead, PullRequestLink, SharedWorktreeRefusal, Todo, @@ -53,6 +52,9 @@ from agent_comms.work_branches import UnpushedHead from conftest import declare_test_bus from source_revision_fixture import SourceRepository +from github_fixture import GitHubFixture + +pytest_plugins = ("github_fixture",) @dataclass @@ -64,7 +66,7 @@ class Fleet: worktrees: Path old_main: str request: str - pulls: dict[int, ListedPullRequest] = field(default_factory=dict) + github: GitHubFixture problems: dict[str, str] = field(default_factory=dict) commander: ClassVar[str] = "lead" @@ -93,21 +95,9 @@ def git(checkout: Path, *arguments: str, at: float | None = None) -> str: timeout=30, ).stdout.strip() - def reading(self) -> type[ProgressRead]: - """The live progress read, answered by this fleet's pull requests instead of GitHub.""" - pulls = self.pulls - - class FleetProgressRead(ProgressRead): - def query( - self, repo: RepositoryName, numbers: tuple[int, ...] - ) -> tuple[ListedPullRequest, ...]: - return tuple(pulls[number] for number in numbers if number in pulls) - - return FleetProgressRead - def observe(self) -> CommanderObservation: return CommanderObservation.read( - self.comms, self.comms.registry.require(self.commander), self.reading() + self.comms, self.comms.registry.require(self.commander) ) def thread(self, name: str) -> Thread: @@ -193,9 +183,10 @@ def claim(self, todo: Todo, number: int, head: str, *, at: float) -> OrderClaim: return claimed def pull_request(self, number: int, branch: str, head: str) -> None: - self.pulls[number] = UnmergedListedPullRequest( - number, branch, head, "2026-10-23T12:00:00Z", "main" - ) + stamp = datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%SZ") + self.github.put(self.repository, UnmergedListedPullRequest( + number, branch, head, stamp, "main", reviewable_times=(stamp,), + )) class IncidentSeed(ABC): @@ -401,7 +392,7 @@ def seed(cls, fleet: Fleet) -> StandingRule: @pytest.fixture -def fleet(comms, tmp_path, configured_repository, installed_runtime): +def fleet(comms, tmp_path, configured_repository, installed_runtime, github_cli): source = SourceRepository.create(tmp_path / "source") declare_test_bus( comms, @@ -413,9 +404,10 @@ def fleet(comms, tmp_path, configured_repository, installed_runtime): request = comms.goals.todos.add_request( comms.registry.require(Fleet.commander).incarnation, "Keep the fleet on its plans" ).id - configured_repository(comms) + project = configured_repository(comms) + github_cli.configure(source.github, *(repo.github for repo in project.repos)) old_main = source.main_commit - built = Fleet(comms, source, tmp_path / "worktrees", old_main, request) + built = Fleet(comms, source, tmp_path / "worktrees", old_main, request, github_cli) # Main moves past the oldest branch's base by exactly one commit more than the limit. for count in range(BranchBehindMain.limit + 1): built.commit(source.checkout, "main.py", f"count = {count}\n") @@ -457,4 +449,6 @@ def test_replay_raises_each_incident_once(fleet): assert refused.value.holding_holder == behind.incarnation second = built.observe() - assert StandingRule.heartbeat_changes(first.rules, second.rules) == () + assert {rule.observation_key() for rule in first.rules} == { + rule.observation_key() for rule in second.rules + } diff --git a/tests/test_head_review.py b/tests/test_head_review.py index 1331fea8f2..1210276fa9 100644 --- a/tests/test_head_review.py +++ b/tests/test_head_review.py @@ -1,46 +1,132 @@ -"""A head verdict goes only on the pull request's current head; a moved head posts nothing.""" +"""Head decisions through the native boundary, real stores, and the external gh process.""" -import subprocess +import time +from dataclasses import replace +from datetime import UTC, datetime import pytest -from agent_comms import repository_name -from agent_comms.errors import ToolRefusedError -from agent_comms.head_review import AcceptedHeadReview, PullRequestHead -from agent_comms.repository_name import RepositoryName +from pathlib import Path -CURRENT = "c" * 40 +from agent_comms.commander import judge, order +from agent_comms.goal_actions import CreateProblemGoalAction +from agent_comms.head_review import PullRequestHead +from agent_comms.recorded_time import RecordedEventTime +from agent_comms.standing_rules import OrderRefusingStandingRule +from agent_comms.errors import RelationViolationError, ToolRefusedError +from agent_comms.tool_caller import ThreadCaller +from agent_comms.order_tools import CommsHeadVerdictTool +from agent_comms.todos import ( + HeadReviewOrder, NoHeadOrder, PullRequestLink, RecordPullRequest, + RejectedHeadVerdict, ReviewedHeadVerdict, StoredHeadReview, StrikeReason, + UnmergedListedPullRequest, +) +pytest_plugins = ("test_holder_standing_rules", "github_fixture") -class Gh: - """Stands in for gh: answers the PR read and records every status post.""" +HEAD = "c" * 40 +REASON = StrikeReason("This head repeats the same decision in two unrelated places instead of using the existing class.") - def __init__(self) -> None: - self.posts: list[tuple[str, ...]] = [] - def __call__(self, command, **_options): - read = command[1] == "api" and command[2].endswith("/pulls/7") - self.posts.extend(() if read else (tuple(command),)) - return subprocess.CompletedProcess(command, 0, f"{CURRENT}\n" if read else "", "") +@pytest.fixture +def head_bus(held_bus, github_cli): + todo = held_bus.goals.todos.get("P2") + github_cli.configure(todo.repo) + stamp = datetime.fromtimestamp(time.time() - 601, UTC).strftime("%Y-%m-%dT%H:%M:%SZ") + github_cli.put(todo.repo, UnmergedListedPullRequest( + 7, "fix/head", HEAD, stamp, "main", reviewable_times=(stamp,), + )) + return held_bus, github_cli, PullRequestHead(todo.repo, 7, HEAD) -@pytest.fixture -def gh(monkeypatch) -> Gh: - stub = Gh() - monkeypatch.setattr(repository_name.subprocess, "run", stub) - return stub +def invoke(comms, request, verdict, order=NoHeadOrder()): + tool = CommsHeadVerdictTool(request=request, operation=verdict, order=order) + return tool.invoke(comms, tool.arguments, caller=ThreadCaller("lead")) + + +def test_orderless_head_family_persists_both_outcomes_and_posts_each_exact_head(head_bus): + comms, github, request = head_bus + before = tuple(todo.strikes for todo in comms.goals.todos.list(request.repo)) + accepted, rejected = ReviewedHeadVerdict(), RejectedHeadVerdict(REASON) + invoke(comms, replace(request, head=HEAD[:9]), accepted) + invoke(comms, request, rejected) + with comms.goals.todos.snapshot() as db: + records = tuple(row.record for row in StoredHeadReview.select(db, order_by=("seq",))) + assert tuple(record.verdict for record in records) == (accepted, rejected) + assert all(record.request == request and record.order == NoHeadOrder() for record in records) + assert comms.goals.todos.head_reviews(request)[0].verdict == rejected + assert tuple(todo.strikes for todo in comms.goals.todos.list(request.repo)) == before + posts = tuple(call for call in github.calls() if f"statuses/{HEAD}" in call[1]) + assert len(posts) == 2 + assert "state=success" in posts[0] and "state=failure" in posts[1] + + +def test_judge_ok_and_back_dispatch_orderless_prs_without_positional_edit_ids(head_bus): + comms, github, request = head_bus + context = order.Commander.read_root(comms.root) + why = "The existing class owns this behavior without another copy" + common = ["--pr", str(request.pr), "--head", request.head, "--why", why] + judge.main(["ok", *common], context) + assert isinstance(comms.goals.todos.head_reviews(request)[0].verdict, ReviewedHeadVerdict) + todo = comms.goals.todos.get("P2") + location = f"{Path(todo.order.check.directory) / todo.order.owner.file}:{todo.order.owner.line}" + judge.main(["back", *common, + "--roast", "This is stupid because the same decision is written twice instead of belonging to its existing class.", + "--fix", "Put the shared behavior on the existing parent and delete both copies.", + "--slop", "IMPL-5", "--fact", "the same decision is written twice", + "--at", location, "--derive", "base-method", "--by", location], context) + assert isinstance(comms.goals.todos.head_reviews(request)[0].verdict, RejectedHeadVerdict) + assert len(tuple(call for call in github.calls() if f"statuses/{HEAD}" in call[1])) == 2 + + +def test_moved_head_posts_and_records_nothing(head_bus): + comms, github, request = head_bus + moved = replace(request, head="a" * 40) + with pytest.raises(ToolRefusedError): + invoke(comms, moved, ReviewedHeadVerdict()) + assert not comms.goals.todos.head_reviews(moved) + assert not any("statuses/" in call[1] for call in github.calls()) -def test_a_moved_head_is_refused_and_posts_nothing(gh: Gh) -> None: - with pytest.raises(ToolRefusedError, match="moved"): - PullRequestHead(RepositoryName.from_wire("org/repo"), 7, "a" * 9).current() - assert gh.posts == [] +def test_ready_orderless_review_refuses_backend_orders_until_a_recorded_decision(head_bus): + comms, _, request = head_bus + todo = comms.goals.todos.get("P2") + assert OrderRefusingStandingRule.refusals(comms.goals, "lead") + action = CreateProblemGoalAction( + problem="P3", commander="lead", request=todo.request, + assigner=comms.registry.require("lead").incarnation, + repo=todo.repo, text="Repair the next owner", order=todo.order.checked(), + review_at=RecordedEventTime(time.time() + 3600), + ) + with pytest.raises(RelationViolationError): + comms.goals.update_goal("outsider", action) + assert all(current.id != "P3" for current in comms.goals.todos.list(request.repo)) + invoke(comms, request, RejectedHeadVerdict(REASON)) + assert not OrderRefusingStandingRule.refusals(comms.goals, "lead") + comms.goals.update_goal("outsider", action) + assert comms.goals.todos.get("P3").holder == "outsider" -def test_the_current_head_is_posted_in_full(gh: Gh) -> None: - repo = RepositoryName.from_wire("org/repo") - head = PullRequestHead(repo, 7, CURRENT[:9]).current() - AcceptedHeadReview().post(repo, head, "claude-ai", "#7") - (post,) = gh.posts - assert f"repos/org/repo/statuses/{CURRENT}" in post - assert "state=success" in post +def test_linked_head_rejection_strikes_revises_and_notifies_its_holder(head_bus): + comms, _, request = head_bus + todo = comms.goals.todos.get("P2") + held = todo.held() + RecordPullRequest(todo.revision, held.owner_identity, + PullRequestLink(request.repo, request.pr), held.generation).record(comms.goals, todo.id) + todo = comms.goals.todos.get(todo.id) + scope = HeadReviewOrder.for_request( + (todo,), request, "Repair the existing owner", (todo.order.checked().owner,), + ) + invoke(comms, request, ReviewedHeadVerdict(), scope) + todo = comms.goals.todos.get(todo.id) + scope = HeadReviewOrder.for_request( + (todo,), request, "Repair the existing owner", (todo.order.checked().owner,), + ) + before = comms.registry.require("worker").goal.require() + invoke(comms, request, RejectedHeadVerdict(REASON), scope) + changed = comms.goals.todos.get(todo.id) + assert changed.strike_count == todo.strike_count + 1 + assert changed.order.checked().id != todo.order.checked().id + assert REASON.text in changed.order.checked().edits + assert comms.registry.require("worker").goal.require().revision > before.revision + assert comms.goals.todos.head_reviews(request)[0].order.problem == todo.id diff --git a/tests/test_held_problem_orders.py b/tests/test_held_problem_orders.py index 95b9f7db96..df9c81ffb0 100644 --- a/tests/test_held_problem_orders.py +++ b/tests/test_held_problem_orders.py @@ -92,13 +92,14 @@ def merge_two_commit_fix(source, binaries, number): source.git(checkout, 'merge', '-q', '--no-ff', '-m', f'Merge pull request #{number}', f'fix/{number}') merge = source.git(checkout, 'rev-parse', 'HEAD').strip() source.git(checkout, 'push', '-q', 'origin', 'main', f'{commits[-1]}:refs/pull/{number}/head') - stub = binaries / 'pull_requests.json' - data = json.loads(stub.read_text()) - data['requests'][str(number)] = {'body': f'## fix {number}', 'listed': FieldCodec.encode( + from github_fixture import GitHubFixture + + GitHubFixture(binaries).put(source.github, MergedListedPullRequest(number=number, branch=f'fix/{number}', head=commits[-1], updated='2026-10-23T12:00:00Z', base=WorkBranches.main, - merge=PullRequestCommit(merge), merged='2026-10-23T12:00:00Z'))} - stub.write_text(json.dumps(data)) + reviewable_times=('2026-10-23T12:00:00Z',), + merge=PullRequestCommit(merge), merged='2026-10-23T12:00:00Z'), + body=f'## fix {number}') return base, commits[0], merge diff --git a/tests/test_holder_standing_rules.py b/tests/test_holder_standing_rules.py index 34047c1d56..0570f687a7 100644 --- a/tests/test_holder_standing_rules.py +++ b/tests/test_holder_standing_rules.py @@ -29,6 +29,8 @@ ) from agent_comms.comms import wire from agent_comms.commander_heartbeat import CommanderHeartbeat +from agent_comms.commander_pings import CommanderUpdate +from agent_comms.service_passes import ServicePassStore from agent_comms.coordinator import Coordination from agent_comms.errors import CommittedMessageError, HumanSendRefusedError, RelationViolationError from agent_comms.field_codec import FieldCodec @@ -649,7 +651,8 @@ def test_startup_heartbeat_enforces_and_publishes_both_new_rule_lines(held_bus, commander = held_bus.registry.require("lead") held_bus.registry.register(replace(commander, process_identity=ProcessIdentity.capture(os.getpid()))) heartbeat.pass_body(held_bus, heartbeat.observe(held_bus)) - assert commander.incarnation in heartbeat.reviews + saved = CommanderUpdate.latest(held_bus, commander.incarnation) + assert saved is not None and saved.handled assert isinstance(held_bus.registry.require("worker").goal.require().state, ActiveGoal) page = MessagePageRequest(DMDisplayScope.capture( heartbeat.sender, "lead", held_bus.registry.snapshot(), @@ -660,81 +663,60 @@ def test_startup_heartbeat_enforces_and_publishes_both_new_rule_lines(held_bus, print(message.body) -def test_every_pass_beats_and_a_stale_beat_is_heartbeat_silent(held_bus, monkeypatch): - """10:08 on 2026-10-21: the heartbeat sent only on change, so 31 failed passes read as calm.""" - from agent_comms.commander_heartbeat import CommanderReview - +def test_successful_pass_health_does_not_depend_on_messages(held_bus, monkeypatch): commander = held_bus.registry.require("lead") held_bus.registry.register(replace(commander, process_identity=ProcessIdentity.capture(os.getpid()))) - commander = held_bus.registry.require("lead") - review = CommanderReview(frozenset()) - CommanderHeartbeat.register(held_bus.registry) - assert HeartbeatSilent.violations(held_bus, "lead") == () - states = UnitState.current( - held_bus, "lead", HeartbeatFrame.read(held_bus, "lead", ProgressRead())) - review.wake(held_bus, CommanderHeartbeat.sender, commander, (), states) - review.wake(held_bus, CommanderHeartbeat.sender, commander, (), states) - page = MessagePageRequest(DMDisplayScope.capture( - CommanderHeartbeat.sender, "lead", held_bus.registry.snapshot(), - )).read_opened(held_bus.bus.bus) - changed, unchanged = (message for message in page.messages - if message.sender == CommanderHeartbeat.sender) - assert "Standing rules clear." in changed.body - assert unchanged.body == f"COMMANDER HEARTBEAT: unchanged since seq {changed.seq}, 0 violations." + passes = ServicePassStore.at(held_bus.root) + passes.passed(CommanderHeartbeat.__name__) + assert not passes.read().commander_completed + worker = held_bus.registry.require("worker").incarnation + observed = (commander.incarnation, worker) + passes.heartbeat_passed((worker,), observed) + passes.heartbeat_passed((commander.incarnation,), observed) + completed = passes.read().commander_completed[commander.incarnation] assert HeartbeatSilent.violations(held_bus, "lead") == () - later = unchanged.timestamp + HeartbeatSilent.seconds + 1 + before = held_bus.bus.bus.revision() + later = completed + HeartbeatSilent.seconds + 1 monkeypatch.setattr(time, "time", lambda: later) - # Neither a commander reply nor a fresh beat sent to another commander proves - # this commander's heartbeat is alive. - held_bus.messaging.send("lead", CommanderHeartbeat.sender, "I read the last beat") - held_bus.messaging.send(CommanderHeartbeat.sender, "outsider", "Another commander is healthy") silent, = HeartbeatSilent.violations(held_bus, "lead") - assert silent.beat == unchanged.seq - assert silent.line().startswith(f"HEARTBEAT SILENT: no beat since seq {unchanged.seq}") + assert isinstance(silent, MissedHeartbeat) + assert silent.since == completed + HeartbeatSilent.seconds + # Another commander's pass cannot clear this alarm. + passes.heartbeat_passed((worker,), observed) + assert HeartbeatSilent.violations(held_bus, "lead") == (silent,) + # A successful quiet pass clears the alarm without inventing a conversation. + passes.heartbeat_passed((commander.incarnation,), observed) + assert HeartbeatSilent.violations(held_bus, "lead") == () + assert held_bus.bus.bus.revision() == before -def test_first_heartbeat_is_owed_from_the_oldest_open_request_then_survives_renames( - held_bus, monkeypatch): +def test_first_pass_is_owed_from_oldest_request_and_survives_renames(held_bus, monkeypatch): commander = held_bus.registry.require("lead") held_bus.registry.register(replace(commander, process_identity=ProcessIdentity.capture(os.getpid()))) request, = CommanderHeartbeat.requested_by(held_bus.registry.require("lead"), held_bus.goals) - assert HeartbeatSilent.violations(held_bus, "lead") == () due = request.created_at + HeartbeatSilent.seconds monkeypatch.setattr(time, "time", lambda: due + 1) never, = HeartbeatSilent.violations(held_bus, "lead") assert isinstance(never, HeartbeatNotStarted) assert never.request == request.id and never.since == due - assert "HEARTBEAT NOT STARTED" in never.line() - assert HeartbeatSilent.broken( - held_bus, "lead", observed=(), frame=HeartbeatFrame.read(held_bus, "lead", ProgressRead()), - ) == (never,) - - # A newer request and mere service registration cannot reset older unpaid service. held_bus.goals.todos.add_request(commander.incarnation, "A newer independent request") CommanderHeartbeat.register(held_bus.registry) + held_bus.messaging.send(CommanderHeartbeat.sender, "lead", "A DM is not a completed pass") assert HeartbeatSilent.violations(held_bus, "lead") == (never,) - fresh = held_bus.bus.publisher.publish_ordinary(Message( - sender=CommanderHeartbeat.sender, target="lead", body="The heartbeat is running now", - type=InfoKind, timestamp=due + 1, - ), destination=CommandTarget.named("lead"), origin=AgentOrigin()) + passes = ServicePassStore.at(held_bus.root) + observed = (commander.incarnation,) + passes.heartbeat_passed(observed, observed) + completed = passes.read().commander_completed[commander.incarnation] assert HeartbeatSilent.violations(held_bus, "lead") == () - held_bus.threads._rename_thread("lead", "renamed-lead") held_bus.threads._rename_thread(CommanderHeartbeat.sender, "renamed-heartbeat") - monkeypatch.setattr(time, "time", lambda: fresh.timestamp + HeartbeatSilent.seconds + 1) - held_bus.messaging.send("renamed-lead", "renamed-heartbeat", "Read the earlier beat") + monkeypatch.setattr(time, "time", lambda: completed + HeartbeatSilent.seconds + 1) silent, = HeartbeatSilent.violations(held_bus, "renamed-lead") - assert isinstance(silent, MissedHeartbeat) and silent.beat == fresh.seq - assert silent.since == fresh.timestamp + HeartbeatSilent.seconds - newer = held_bus.bus.publisher.publish_ordinary(Message( - sender="renamed-heartbeat", target="renamed-lead", body="Heartbeat after both renames", - type=InfoKind, timestamp=time.time(), - ), destination=CommandTarget.named("renamed-lead"), origin=AgentOrigin()) + assert isinstance(silent, MissedHeartbeat) + assert silent.since == completed + HeartbeatSilent.seconds + renamed = (held_bus.registry.require("renamed-lead").incarnation,) + passes.heartbeat_passed(renamed, renamed) assert HeartbeatSilent.violations(held_bus, "renamed-lead") == () - monkeypatch.setattr(time, "time", lambda: newer.timestamp + HeartbeatSilent.seconds + 1) - held_bus.messaging.send("renamed-lead", "renamed-heartbeat", "Read the new beat too") - latest, = HeartbeatSilent.violations(held_bus, "renamed-lead") - assert isinstance(latest, MissedHeartbeat) and latest.beat == newer.seq def test_one_answer_covers_an_escalation_burst_sent_before_it(held_bus): diff --git a/tests/test_proof_build.py b/tests/test_proof_build.py index 43c521a736..1225406607 100644 --- a/tests/test_proof_build.py +++ b/tests/test_proof_build.py @@ -324,8 +324,6 @@ def test_build_containment_does_not_turn_git_failure_into_absence_or_success(pro def test_install_due_compares_fetched_main_to_installed_build( tmp_path, proof_build_checkouts, advance_main, monkeypatch, ): - # The settle window is its own test; here every uninstalled merge has settled. - monkeypatch.setattr(InstallDue, "settle", 0) graph = next(graph for graph in proof_build_checkouts if graph.repository.github == CorePackage.repository) checkout = graph.repository.checkout @@ -344,19 +342,18 @@ def test_install_due_compares_fetched_main_to_installed_build( bus.registry.declare(Thread("lead", frozenset(), str(checkout))) Commander.on_root("lead", (graph.repository,), bus.root) + after_settle = time.time() + InstallDue.settle + 1 + monkeypatch.setattr(time, "time", lambda: after_settle) due = InstallDue.violations(bus, "lead") assert Repo.git(checkout, "rev-parse", "HEAD") == before assert Repo.git(checkout, "rev-parse", "origin/main").strip() == ( graph.newer if advance_main else graph.installed) assert len(due) == int(advance_main) - changes = StandingRule.heartbeat_changes((), due) - assert len(changes) == int(advance_main) for alarm in due: - assert graph.repository.github.to_wire() in alarm.why - assert graph.newer in alarm.why - assert graph.installed in alarm.why - assert alarm.line() in changes[0] + assert alarm.repo == graph.repository.github + assert alarm.main == graph.newer + assert alarm.installed == graph.installed # An install advances the canonical manifest, not a clock or bus schedule. runtime = LaunchRuntime.installed().prefix @@ -364,7 +361,6 @@ def test_install_due_compares_fetched_main_to_installed_build( RuntimeResolution(core_commit=graph.newer, packages=resolution.packages).save(runtime) after_install = InstallDue.violations(bus, "lead") assert after_install == () - assert len(StandingRule.heartbeat_changes(due, after_install)) == int(advance_main) assert Repo.git(checkout, "rev-parse", "HEAD") == before @@ -390,6 +386,6 @@ def test_install_waits_for_merges_to_settle_into_one_batch( monkeypatch.setattr(time, "time", lambda: merged + InstallDue.settle) (due,) = InstallDue.violations(bus, "lead") - assert graph.newer in due.why - assert "oldest 30 min" in due.why + assert due.main == graph.newer + assert due.oldest_minutes == 30 diff --git a/tests/test_storage_violations.py b/tests/test_storage_violations.py index 4074c514b4..9c875e2512 100644 --- a/tests/test_storage_violations.py +++ b/tests/test_storage_violations.py @@ -167,4 +167,4 @@ def held_read(code: CodeType, _offset: int) -> None: assert str(home) in text assert str(comms.registry.storage_rule.order_checkout(worker.name, "P")) in text assert str(worker.storage_roots.scratch) in text - print("\n".join(StandingRule.heartbeat_changes((), found))) + print("\n".join(rule.line() for rule in found)) diff --git a/tests/test_todos.py b/tests/test_todos.py index ec25184649..0baac908bd 100644 --- a/tests/test_todos.py +++ b/tests/test_todos.py @@ -73,7 +73,7 @@ from agent_comms.work_branches import WorkBranches from source_revision_fixture import SourceRepository -pytest_plugins = ("test_holder_standing_rules", "test_unowned_checkout_work", "test_proof_build") +pytest_plugins = ("test_holder_standing_rules", "test_unowned_checkout_work", "test_proof_build", "github_fixture") def thread(name: str, root: Path, created: float) -> Thread: @@ -272,7 +272,7 @@ def test_merged_order_keeps_its_checked_fields(tmp_path: Path, source_repository @pytest.fixture -def pr_commander(tmp_path, monkeypatch, proof_build_checkouts, durable_home): +def pr_commander(tmp_path, monkeypatch, proof_build_checkouts, durable_home, github_cli): """One cited source, its real bare remote and installed history; only GitHub is synthetic.""" graph = next(graph for graph in proof_build_checkouts if graph.repository.github == CorePackage.repository) @@ -290,7 +290,7 @@ def pr_commander(tmp_path, monkeypatch, proof_build_checkouts, durable_home): for number in (1, 2): source.git(source.checkout, 'push', 'origin', f'{merged}:refs/pull/{number}/head') common = dict(branch="feature/orders", head=merged, updated="2026-10-20T12:00:00Z", - base=WorkBranches.main) + base=WorkBranches.main, reviewable_times=("2026-10-20T12:00:00Z",)) requests = ( (MergedListedPullRequest(number=1, merge=PullRequestCommit(merged), merged="2026-10-20T12:00:00Z", **common), "No order metadata."), @@ -298,15 +298,9 @@ def pr_commander(tmp_path, monkeypatch, proof_build_checkouts, durable_home): merged="2026-10-20T12:00:00Z", **common), "## P343\n\n## P344"), (ClosedListedPullRequest(number=3, **common), "Closed without merging."), ) - binaries = tmp_path / "bin" - binaries.mkdir() - (binaries / "gh").symlink_to(Path(__file__).with_name("github_cli_stub.py")) - (binaries / "pull_requests.json").write_text(json.dumps({ - "repo": source.github.to_wire(), - "requests": {str(row.number): {"body": body, "listed": FieldCodec.encode(row)} - for row, body in requests}, - })) - monkeypatch.setenv("PATH", os.pathsep.join((str(binaries), os.environ["PATH"]))) + github_cli.configure(source.github) + for row, body in requests: + github_cli.put(source.github, row, body) Bus.install(Bus.path_in(tmp_path)) lead = Thread( 'lead', frozenset(), str(source.checkout), created_at=1.0) diff --git a/tests/test_unowned_checkout_work.py b/tests/test_unowned_checkout_work.py index 01e9503f2b..d2eb5c5f6f 100644 --- a/tests/test_unowned_checkout_work.py +++ b/tests/test_unowned_checkout_work.py @@ -25,8 +25,8 @@ from agent_comms.commander_heartbeat import ( CommanderHeartbeat, CommanderObservation, - CommanderReview, ) +from agent_comms.commander_pings import CommanderUpdate, RulePing from agent_comms.comms import Comms, wire from agent_comms.field_codec import FieldCodec from agent_comms.goals import OrderCheck @@ -202,11 +202,9 @@ def test_dirty_then_unpushed_then_pushed_linked_checkout(checkout_bus: Comms, tm practice.include(tuple(RootFile.existing_in(practice.root))) (root, _measurement) = practice.path, practice.measurement c = wire(root) - review = CommanderReview(frozenset()) - def beat(): frame = CommanderObservation.read(c, c.registry.require('lead')) - review.wake(c, CommanderHeartbeat.sender, frame.thread, frame.rules, frame.states) + CommanderHeartbeat().pass_body(c, (frame,)) with LaunchRoot.from_path(root).selected(): (checkout / 'work.py').write_text('# not yet published\n') @@ -229,7 +227,9 @@ def beat(): assert len(available) == 1 assert isinstance(available[0], ReassignedWheelSpin) assert available[0].unit == 'idle' - assert set(available).isdisjoint(review.rules) + previous = CommanderUpdate.latest(c, c.registry.require('lead').incarnation) + assert previous is not None + assert set(map(RulePing.capture, available)).isdisjoint(previous.handled) UnownedOrderWatch().watch_pass(lambda: nullcontext(c)) assigned = c.goals.todos.get('P2') assert assigned.holder == 'idle' diff --git a/tools/cutover/carry_head_reviews.py b/tools/cutover/carry_head_reviews.py new file mode 100644 index 0000000000..5f4e614d77 --- /dev/null +++ b/tools/cutover/carry_head_reviews.py @@ -0,0 +1,110 @@ +"""Carry embedded order-head reviews before the install reconciles TodoStore. + +The insert and removal share one transaction, so a completed carry has no old +review left to carry again. Delete this step after the live installation. +""" + +from __future__ import annotations + +import json +from contextlib import closing +from dataclasses import dataclass +from typing import Annotated, ClassVar, Literal + +from pydantic import JsonValue + +from agent_comms.field_codec import FieldCodec, GitCommitText +from agent_comms.head_review import PullRequestHead +from agent_comms.installer.probes.carry import Carried, Carry +from agent_comms.locked_store import RootFile +from agent_comms.maintenance_barrier import MaintenanceBarrier, MaintenanceHolder +from agent_comms.recorded_time import EventTime +from agent_comms.registry_store import RegistryStore +from agent_comms.repository_name import RepositoryName +from agent_comms.thread_identity import ThreadIncarnation +from agent_comms.todos import ( + Assignment, + CheckedProblemOrder, + HeadReviewRecord, + ProblemHeadReview, + ProblemOrder, + ReviewedHeadVerdict, + StoredHeadReview, + Todo, + TodoStore, +) +from agent_comms.typed_table import Column, TextStorage, TypedRow + + +@dataclass(frozen=True) +class EmbeddedHeadReview: + kind: Literal["reviewed_head_verdict"] + expected_revision: int + actor: ThreadIncarnation + previous: Assignment + at: EventTime + quote: str + head: Annotated[str, GitCommitText] + + def record(self, request: PullRequestHead, problem: str) -> HeadReviewRecord: + return HeadReviewRecord( + request, self.actor, self.at, ReviewedHeadVerdict(), + ProblemHeadReview(quote=self.quote, problem=problem, previous=self.previous), + ) + + +@dataclass(frozen=True) +class EmbeddedReviewSource(TypedRow): + id: str + repo: Annotated[RepositoryName, Column(storage=TextStorage)] + order: ProblemOrder + prior_orders: tuple[CheckedProblemOrder, ...] + verdicts: tuple[dict[str, JsonValue], ...] + + def records(self) -> tuple[HeadReviewRecord, ...]: + links = {link for order in (*self.prior_orders, *self.order.declarations) + for link in order.pull_requests} + if len(links) != 1: + raise ValueError(f"{self.id}: embedded head review needs one recorded PR, got {links}") + link, = links + if link.repo != self.repo: + raise ValueError(f"{self.id}: embedded head review names another repository") + reviews = (FieldCodec.decode(EmbeddedHeadReview, wire) for wire in self.verdicts + if wire["kind"] == "reviewed_head_verdict") + return tuple(review.record(PullRequestHead(self.repo, link.number, review.head), self.id) + for review in reviews) + + def remaining(self) -> tuple[dict[str, JsonValue], ...]: + return tuple(wire for wire in self.verdicts if wire["kind"] != "reviewed_head_verdict") + + +class HeadReviews(Carry): + opened_files: ClassVar[tuple[type[RootFile], ...]] = (TodoStore,) + + def carry(self, *, apply: bool) -> Carried: + path = TodoStore.path_in(self.root) + self.require_confined(self.root, path) + if not path.exists(): + return Carried(0) + if apply: + MaintenanceBarrier(RegistryStore.path_in(self.root)).read().require_held_by( + MaintenanceHolder.current().nonce, "Carry embedded PR-head reviews") + with TodoStore.observing(path) as db: + sources = EmbeddedReviewSource.read(db.execute( + f'SELECT id,repo,"order",prior_orders,verdicts FROM "{Todo.declared_name}" ' + "WHERE EXISTS (SELECT 1 FROM json_each(verdicts) " + "WHERE json_extract(value,'$.kind')='reviewed_head_verdict')")) + prepared = tuple((source, source.records()) for source in sources) + count = sum(len(records) for _, records in prepared) + if apply and count: + with closing(TodoStore.writer(path)) as db, db: + db.execute("PRAGMA foreign_keys=ON") + db.execute("BEGIN IMMEDIATE") + StoredHeadReview.create(db, if_not_exists=True) + for source, records in prepared: + for record in records: + StoredHeadReview.insert_values(db, record=record) + db.execute(f'UPDATE "{Todo.declared_name}" SET verdicts=? WHERE id=?', + (json.dumps(source.remaining()), source.id)) + return Carried(count, detail={"reviews": count, "todos": len(prepared)}, + rewritten=(path.relative_to(self.root),) if apply and count else ())