diff --git a/src/agent_comms/commander/unit_talk.py b/src/agent_comms/commander/unit_talk.py new file mode 100644 index 0000000000..a2aee0ef3b --- /dev/null +++ b/src/agent_comms/commander/unit_talk.py @@ -0,0 +1,188 @@ +"""Standing rules on what a commanded unit says, read from its own published messages. + +Waste has to flag itself: a unit that announces a wait nobody owns, or talks far more than +it commits, gets a named board row without anyone reading #comms by hand. The rule text these +rows enforce is rules/no-blocking.md. +""" + +from __future__ import annotations + +import re +from abc import abstractmethod +from collections.abc import Iterator, Sequence +from dataclasses import dataclass, field +from functools import cached_property +from typing import TYPE_CHECKING, ClassVar, Self + +from agent_comms.comms import Comms +from agent_comms.field_codec import MessageId +from agent_comms.standing_rules import ( + DatedStandingRule, + FamilyScan, + HolderLine, + UndatedStandingRule, +) +from agent_comms.thread_identity import ThreadIncarnation + +if TYPE_CHECKING: + from agent_comms.bus_publication import Delivered + from agent_comms.heartbeat_frame import HeartbeatFrame + from agent_comms.ui_model.unit_state import UnitObservation, UnitState + + +@dataclass(frozen=True) +class UnitTalk: + """One observed unit's own messages and commits over the rules' shared window.""" + + source: UnitObservation + window: ClassVar[float] = 3600 + + @property + def unit(self) -> ThreadIncarnation: + return self.source.thread.incarnation + + @property + def since(self) -> float: + return self.source.now - self.window + + @cached_property + def messages(self) -> tuple[Delivered, ...]: + """Its own publications in the window, newest first, through the bus's sent index.""" + return tuple( + message + for record in self.source.comms.bus.bus.sent( + self.unit.require_recorded().lookup, limit=500 + ) + for message in record.deliveries() + if message.timestamp >= self.since + ) + + @cached_property + def commits(self) -> int: + """Distinct commits its held orders' sources recorded in the window.""" + return len( + { + point + for origin in self.source.sources + for point in origin.commits() + if point.at >= self.since + } + ) + + @property + def characters(self) -> int: + return sum(message.text_size for message in self.messages) + + +class UnitTalkRule(HolderLine, FamilyScan, declared_name=None): + """Rules over one unit's talk share the heartbeat's existing unit observations.""" + + @classmethod + def scan_observed( + cls, comms: Comms, commander: str, observed: Sequence[UnitState], frame: HeartbeatFrame + ) -> tuple[Self, ...]: + talks = tuple(UnitTalk(state.source) for state in observed) + return tuple( + rule + for member in cls.members_with(cls) + for talk in talks + for rule in member.derive(talk) + ) + + @classmethod + @abstractmethod + def derive(cls, talk: UnitTalk) -> Iterator[Self]: ... + + +@dataclass(frozen=True) +class BlockedTalk(UnitTalkRule, DatedStandingRule): + """A unit says it is waiting, frozen or holding and names neither an owner nor a line. + + no-blocking.md: every wait names its owner and is announced; a wait nobody can act on is + a deadlock nobody can see. A named thread or a file:line is what makes it actionable.""" + + holder: ThreadIncarnation + message_id: MessageId + at: float = field(compare=False) + quote: str = field(compare=False) + words: ClassVar[tuple[str, ...]] = ( + "waiting on", + "waiting for", + "waits on", + "awaiting", + "blocked on", + "blocked by", + "blocked until", + "frozen until", + "frozen for", + "head frozen", + "holding until", + "holding for", + "on hold", + "deferred to", + "deferring to", + "parked until", + "until root", + "pending root", + "pending review", + ) + """Phrases that announce a stop, not words that merely occur in one: "freeze is + superseded" or "preserving behaviour" stop nothing.""" + stall: ClassVar[re.Pattern[str]] = re.compile( + r"(? Iterator[Self]: + for message in talk.messages: + said = cls.stall.search(message.body) + if said and not ( + cls.located.search(message.body) + or cls.named_owner.search(message.body) + or message.mentions + ): + start = max(0, said.start() - 60) + yield cls( + talk.unit, + message.message_id, + message.timestamp, + " ".join(message.body[start : said.end() + 60].split()), + ) + + def began(self) -> float: + return self.at + + def line(self) -> str: + return ( + f'BLOCKED-TALK {self.holder.name} said "…{self.quote}…" ({self.message_id})' + " with no owner and no file:line. waiting on whom, exactly? -> name the owner" + " and the file:line, or do the next step now" + ) + + +@dataclass(frozen=True) +class Ceremony(UnitTalkRule, UndatedStandingRule): + """A unit publishes far more text than it commits: the report has become the work.""" + + holder: ThreadIncarnation + characters: int = field(compare=False) + commits: int = field(compare=False) + per_commit: ClassVar[int] = 6000 + + @classmethod + def derive(cls, talk: UnitTalk) -> Iterator[Self]: + # Only a unit holding pending work owes commits; an idle or review-only unit does not. + if talk.source.pending and talk.characters > cls.per_commit * max(talk.commits, 1): + yield cls(talk.unit, talk.characters, talk.commits) + + def line(self) -> str: + return ( + f"CEREMONY {self.holder.name} published {self.characters} characters and" + f" {self.commits} commits in the last hour, limit {self.per_commit} per commit." + " the receipt printer is louder than the compiler -> commit, diff stat," + " file:line, Assumed:, nothing else" + ) diff --git a/src/agent_comms/plan_phases.py b/src/agent_comms/plan_phases.py index 338e1d3b8a..78281e2f69 100644 --- a/src/agent_comms/plan_phases.py +++ b/src/agent_comms/plan_phases.py @@ -917,7 +917,7 @@ class DriftOverlapRulesPhase(FleetAwarenessPhase): evidence = AllEvidence( ( ClassDeclared("BranchBehindMain", ("src/agent_comms",)), - ClassDeclared("SharedFileBranches", ("src/agent_comms",)), + ClassDeclared("ConflictingBranches", ("src/agent_comms",)), ) ) diff --git a/src/agent_comms/repo.py b/src/agent_comms/repo.py index 15eb268439..27737637e3 100644 --- a/src/agent_comms/repo.py +++ b/src/agent_comms/repo.py @@ -298,6 +298,18 @@ def grep(cls, checkout: Path | str, *args: str) -> tuple[str, ...]: raise FailedGitCommand(str(checkout), run) return tuple(run.stdout.decode().splitlines()) + @classmethod + def merge_conflicts(cls, checkout: Path | str, left: str, right: str) -> tuple[str, ...]: + """The paths ``git merge-tree`` cannot merge between two commits: exit 0 merges + cleanly, exit 1 lists the conflicting paths after the written tree.""" + run = cls._run(checkout, "merge-tree", "--write-tree", "--name-only", "--no-messages", + left, right) + if run.returncode == 1: + return tuple(run.stdout.decode().splitlines()[1:]) + if run.returncode: + raise FailedGitCommand(str(checkout), run) + return () + @classmethod def of(cls, checkout: Path) -> Repo: """Its GitHub name comes from the checkout's own origin remote.""" diff --git a/src/agent_comms/standing_rules.py b/src/agent_comms/standing_rules.py index adedd44ebb..b8073b43ef 100644 --- a/src/agent_comms/standing_rules.py +++ b/src/agent_comms/standing_rules.py @@ -10,7 +10,7 @@ import time from abc import abstractmethod from collections import Counter -from collections.abc import Iterable, Mapping, Sequence +from collections.abc import Iterable, Iterator, Mapping, Sequence from dataclasses import dataclass, field, fields from functools import cache from itertools import combinations @@ -410,8 +410,39 @@ def line(self) -> str: " who took them, and why? -> restore its required tools") +class HolderBranch: + """A holder's branch difference; two of them may write the same files in parallel.""" + + holder: ThreadIncarnation + difference: BranchDifference + + @staticmethod + def touching[Branch: HolderBranch]( + branches: Sequence[Branch], + ) -> Iterator[tuple[Branch, Branch]]: + """Each pair of branches writing a common file of one repository, once. Indexing by + (repository, file) selects them; any other pair cannot conflict, and a pair on one + branch or under one holder that does conflict is still a conflict to resolve.""" + sharing: dict[tuple[RepositoryName, str], list[int]] = {} + for position, branch in enumerate(branches): + for path in branch.difference.files: + sharing.setdefault((branch.difference.repository, path), []).append(position) + pairs = dict.fromkeys(pair for group in sharing.values() for pair in combinations(group, 2)) + return ((branches[left], branches[right]) for left, right in pairs) + + +@dataclass(frozen=True) +class PreparedBranch(HolderBranch): + """One held order's captured branch difference, before the cut, in its repository.""" + + holder: ThreadIncarnation + todo: Todo + difference: BranchDifference + repository: Path + + @dataclass(frozen=True) -class HeldBranch: +class HeldBranch(HolderBranch): """One order still writing source, joined to its own allocated checkout. Only the branch difference and publication are captured from that worktree before observation; its unit may retire it mid-frame. Later PR comparisons use the order's repository, which outlives @@ -454,6 +485,7 @@ class HeldBranchRule(HolderLine, UndatedStandingRule, FamilyScan, declared_name= def prepare(cls, comms: Comms, commander: str, holds: HeldOrders, pulls: ProgressRead) -> None: storage = comms.registry.storage_rule + prepared: list[PreparedBranch] = [] for held in holds.live: todo = held.todo directory = todo.order.checked().check.directory @@ -464,10 +496,18 @@ def prepare(cls, comms: Comms, commander: str, holds: HeldOrders, path = storage.order_checkout(storage.roots_of(held.holder), todo.id) if path.is_dir(): pulls.checkout(path).branches.prepare_upstreams() - pulls.branch_difference(path) + prepared.append(PreparedBranch(held.holder, todo, pulls.branch_difference(path), + pulls.checkout(directory).path)) except (OSError, GitCommandError, WorkBranchError, TimeoutExpired) as error: pulls.unread(todo, error) + for member in cls.members_with(cls): + member.prepare_captured(tuple(prepared), pulls) + + @classmethod + def prepare_captured(cls, prepared: tuple[PreparedBranch, ...], pulls: ProgressRead) -> None: + """A member that needs more than each branch's own difference reads it here, before + the cut; its scan consumes only what this captured.""" @classmethod def scan_observed(cls, comms: Comms, commander: str, @@ -554,8 +594,12 @@ def line(self) -> str: @dataclass(frozen=True) -class SharedFileBranches(HeldBranchRule): - """Different held branches touch the same repository-relative file.""" +class ConflictingBranches(HeldBranchRule): + """Two held branches that Git cannot merge: one row per pair, only for a real conflict. + + Two branches touching the same file is ordinary parallel work; the row exists only + when ``git merge-tree`` of their heads conflicts, and merging one into the other clears + it, so no row asks a question nothing can answer.""" problem: str holder: ThreadIncarnation @@ -564,28 +608,33 @@ class SharedFileBranches(HeldBranchRule): other_holder: ThreadIncarnation other_branch: str repository: RepositoryName - path: str + paths: tuple[str, ...] = field(compare=False) + + @classmethod + def prepare_captured(cls, prepared: tuple[PreparedBranch, ...], pulls: ProgressRead) -> None: + for first, second in HolderBranch.touching(prepared): + pulls.branch_conflicts(first.repository, first.difference.head, second.difference.head) @classmethod def derive(cls, branches: tuple[HeldBranch, ...]) -> tuple[Self, ...]: - found = [] - for first, second in combinations(branches, 2): - left, right = first.difference, second.difference - if (first.holder == second.holder or left.repository != right.repository - or left.branch == right.branch): - continue - found.extend(cls(problem=first.problem, holder=first.holder, branch=left.branch, - other_problem=second.problem, other_holder=second.holder, - other_branch=right.branch, repository=left.repository, path=path) - for path in sorted(left.files & right.files)) - return tuple(found) + return tuple( + cls(problem=first.problem, holder=first.holder, branch=first.difference.branch, + other_problem=second.problem, other_holder=second.holder, + other_branch=second.difference.branch, repository=first.difference.repository, + paths=conflicts) + for first, second in HolderBranch.touching(branches) + if (conflicts := first.source.pulls.require_branch_conflicts( + first.difference.head, second.difference.head)) + ) def line(self) -> str: - return (f"SEQUENCE {self.holder.name} {self.problem} and {self.other_holder.name} {self.other_problem}" - f" both touch {self.repository.to_wire()}:{self.path}. who goes first? -> sequence" - " them") + return (f"CONFLICT {self.holder.name} {self.problem} ({self.branch}) and" + f" {self.other_holder.name} {self.other_problem} ({self.other_branch}) do not merge" + f" in {self.repository.to_wire()}: {', '.join(self.paths)}. -> merge one into the" + " other and resolve") +@dataclass(frozen=True) class DatedStandingRule(StandingRule, declared_name=None): """A rule that says when its condition began: it is broken on the reads after that.""" diff --git a/src/agent_comms/todos.py b/src/agent_comms/todos.py index b4fd33abcc..a3e7130b1f 100644 --- a/src/agent_comms/todos.py +++ b/src/agent_comms/todos.py @@ -1504,6 +1504,7 @@ def __init__(self) -> None: self._checkouts: dict[Path, PresentCheckout] = {} self._branch_mains: dict[RepositoryName, str] = {} self._branch_facts: dict[Path, BranchDifference] = {} + self._conflicts: dict[tuple[str, str], tuple[str, ...]] = {} def checkout(self, path: Path) -> PresentCheckout: """This read's one reader of the checkout at path.""" @@ -1540,6 +1541,21 @@ def require_branch_difference(self, path: Path) -> BranchDifference: raise WorkBranchError(f"Unprepared branch difference in {path}") return self._branch_facts[path] + def branch_conflicts(self, checkout: Path, left: str, right: str) -> tuple[str, ...]: + """The paths two commits cannot merge, read once per pair in this cut; a merge has + no direction, so either order names the same answer.""" + pair = (min(left, right), max(left, right)) + if pair not in self._conflicts: + self._conflicts[pair] = Repo.merge_conflicts(checkout, *pair) + return self._conflicts[pair] + + def require_branch_conflicts(self, left: str, right: str) -> tuple[str, ...]: + """Consume this cut's prepared merge answer without opening another Git read.""" + pair = (min(left, right), max(left, right)) + if pair not in self._conflicts: + raise WorkBranchError(f"Unprepared merge of {left} and {right}") + return self._conflicts[pair] + class ProgressRead(PullRequests, CheckoutReads): """One read of order progress: its GitHub facts and its checkouts' Git facts.""" diff --git a/tests/test_fleet_replay.py b/tests/test_fleet_replay.py index aa0935fbb1..a32804568c 100644 --- a/tests/test_fleet_replay.py +++ b/tests/test_fleet_replay.py @@ -39,10 +39,10 @@ from agent_comms.standing_rules import ( BranchBehindMain, BranchGrowth, + ConflictingBranches, DeliveredHeadReview, HeldBranchRule, LostTools, - SharedFileBranches, StandingRule, ) from agent_comms.thread_identity import ReportedTools, ToolName @@ -245,8 +245,8 @@ def seed(cls, fleet: Fleet) -> StandingRule: ) -class SharedFileBranchesSeed(IncidentSeed): - rule = SharedFileBranches +class ConflictingBranchesSeed(IncidentSeed): + rule = ConflictingBranches @classmethod def seed(cls, fleet: Fleet) -> StandingRule: @@ -257,7 +257,7 @@ def seed(cls, fleet: Fleet) -> StandingRule: fleet.push(worktree) held.append(fleet.unit(branch, worktree)) left, right = held - return SharedFileBranches( + return ConflictingBranches( left.id, left.held().owner_identity, "shared-left", @@ -265,7 +265,7 @@ def seed(cls, fleet: Fleet) -> StandingRule: right.held().owner_identity, "shared-right", fleet.repository, - "shared.py", + ("shared.py",), ) diff --git a/tests/test_unit_talk.py b/tests/test_unit_talk.py new file mode 100644 index 0000000000..fe2b7457e8 --- /dev/null +++ b/tests/test_unit_talk.py @@ -0,0 +1,67 @@ +from dataclasses import dataclass, field +from types import SimpleNamespace + +import pytest + +from agent_comms.commander.unit_talk import BlockedTalk, Ceremony, UnitTalk +from agent_comms.field_codec import MessageId +from agent_comms.thread_identity import RecordedThreadIncarnation + +UNIT = RecordedThreadIncarnation("install-speed", 1792869186.2298183) + + +@dataclass(frozen=True) +class Talk(UnitTalk): + """The real window over fixed messages and commits instead of a live bus.""" + + said: tuple[SimpleNamespace, ...] = () + committed: int = 0 + holding: bool = True + source: SimpleNamespace = field(default_factory=lambda: SimpleNamespace( + thread=SimpleNamespace(incarnation=UNIT), now=0.0, pending=(), sources=())) + + @property + def messages(self): + return self.said + + @property + def commits(self) -> int: + return self.committed + + def __post_init__(self) -> None: + self.source.pending = ("P964",) if self.holding else () + + +def said(body: str, mentions: tuple = ()) -> SimpleNamespace: + return SimpleNamespace(body=body, text_size=len(body), timestamp=10.0, mentions=mentions, + message_id=MessageId("0123456789ab")) + + +@pytest.mark.parametrize("body", ( + "Source delivered; awaiting root review.", + "Published pair frozen for your review.", + "I'm still waiting on the judge's verdict for #7.", +)) +def test_an_unowned_wait_is_a_named_row(body: str) -> None: + row, = BlockedTalk.derive(Talk(said=(said(body),))) + assert row.holder == UNIT and row.message_id == MessageId("0123456789ab") + + +@pytest.mark.parametrize("message", ( + said("waiting on @codex-greentext to record the batch"), + said("blocked on src/agent_comms/todos.py:4114 refusing the batch"), + said("not waiting on this conflict; continuing"), + said("Origin wasn't waiting for a roll call."), + said("preserving the existing behaviour; the author freeze is superseded"), + said("waiting for the reviewer", mentions=("root",)), +)) +def test_an_owned_negated_or_ordinary_message_is_not(message: SimpleNamespace) -> None: + assert tuple(BlockedTalk.derive(Talk(said=(message,)))) == () + + +def test_ceremony_counts_characters_per_commit_while_holding_work() -> None: + loud = (said("x" * (Ceremony.per_commit * 2 + 1)),) + row, = Ceremony.derive(Talk(said=loud, committed=2)) + assert (row.characters, row.commits) == (Ceremony.per_commit * 2 + 1, 2) + assert tuple(Ceremony.derive(Talk(said=loud, committed=3))) == () + assert tuple(Ceremony.derive(Talk(said=loud, committed=0, holding=False))) == ()