From afd5c07efeb708d76d5fbe2e4673beeb0cbc47dd Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Thu, 29 Oct 2026 18:43:51 -0400 Subject: [PATCH 1/3] Flag stalls and ceremony on the board, and sequence only branches that conflict SharedFileBranches emitted one SEQUENCE row per shared file between two held orders and no answer could clear it: 12 such rows on tonight's board asked who goes first about pairs that merge cleanly. ConflictingBranches emits one row per order pair, only when git merge-tree of the two heads conflicts, naming the conflicting paths; merging one into the other clears it. Repo.merge_conflicts owns the merge-tree exit contract. BlockedTalk (rules/no-blocking.md) names a unit whose message announces a stop (waiting on, awaiting, frozen for, until root...) with neither an owner nor a file:line. Ceremony names a unit holding pending work that published more than 6000 characters per commit in the last hour. Both read the heartbeat's existing unit observations: the bus sent index and the held orders' progress sources. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01RnH6dR85kyFKcc2tQAJLhK --- src/agent_comms/commander/unit_talk.py | 147 +++++++++++++++++++++++++ src/agent_comms/plan_phases.py | 2 +- src/agent_comms/repo.py | 12 ++ src/agent_comms/standing_rules.py | 46 +++++--- tests/test_fleet_replay.py | 10 +- tests/test_unit_talk.py | 67 +++++++++++ 6 files changed, 260 insertions(+), 24 deletions(-) create mode 100644 src/agent_comms/commander/unit_talk.py create mode 100644 tests/test_unit_talk.py diff --git a/src/agent_comms/commander/unit_talk.py b/src/agent_comms/commander/unit_talk.py new file mode 100644 index 0000000000..cce52ddfe0 --- /dev/null +++ b/src/agent_comms/commander/unit_talk.py @@ -0,0 +1,147 @@ +"""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 message in self.source.comms.bus.sent( + self.unit.require_recorded().lookup, limit=500) + 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..18b8240954 100644 --- a/src/agent_comms/standing_rules.py +++ b/src/agent_comms/standing_rules.py @@ -34,7 +34,7 @@ from agent_comms.owner_requests import OwnerRequestId, RequestDirective from agent_comms.recorded_time import EventTime, RecordedEventTime from agent_comms.repository_name import RepositoryName -from agent_comms.repo import GitCommandError +from agent_comms.repo import GitCommandError, Repo from agent_comms.service_passes import ServicePassStore from agent_comms.storage_rules import MemoryUse, StorageRule, UnitStorage from agent_comms.child_process import ProcessIdentity, ResidentMemory @@ -428,6 +428,12 @@ class HeldBranch: def problem(self) -> str: return self.todo.id + def parallel_to(self, other: HeldBranch) -> bool: + """Different holders writing different branches of one repository with shared files.""" + left, right = self.difference, other.difference + return (self.holder != other.holder and left.repository == right.repository + and left.branch != right.branch and bool(left.files & right.files)) + class HolderLine(StandingRule, declared_name=None): """A rule about one holder's order: a recorded action reaching the holder disposes its line.""" @@ -554,8 +560,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 +574,28 @@ class SharedFileBranches(HeldBranchRule): other_holder: ThreadIncarnation other_branch: str repository: RepositoryName - path: str + paths: tuple[str, ...] = field(compare=False) @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 combinations(branches, 2) + if first.parallel_to(second) and (conflicts := Repo.merge_conflicts( + first.repository.path, 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/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))) == () From 824018afc5ecc03bacfdd4f9ed741679fd3f7d41 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Thu, 29 Oct 2026 18:53:06 -0400 Subject: [PATCH 2/3] Capture branch conflicts before the cut; read sent messages through Bus The scan must consume only what prepare captured: running merge-tree in derive opened Git after the cut and broke the complete-cut guarantee. HeldBranchRule.prepare now hands each member the prepared branches; ConflictingBranches reads merge-tree there into ProgressRead's cut (keyed by the unordered commit pair) and derive requires the captured answer. HolderBranch owns the parallel-pair test for both the prepared and the held branch. UnitTalk reads the sent index on comms.bus.bus, where Bus owns it, and unpacks each record's deliveries. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01RnH6dR85kyFKcc2tQAJLhK --- src/agent_comms/commander/unit_talk.py | 4 +- src/agent_comms/standing_rules.py | 55 ++++++++++++++++++++------ src/agent_comms/todos.py | 16 ++++++++ 3 files changed, 62 insertions(+), 13 deletions(-) diff --git a/src/agent_comms/commander/unit_talk.py b/src/agent_comms/commander/unit_talk.py index cce52ddfe0..2cb7553284 100644 --- a/src/agent_comms/commander/unit_talk.py +++ b/src/agent_comms/commander/unit_talk.py @@ -49,9 +49,9 @@ def since(self) -> float: def messages(self) -> tuple[Delivered, ...]: """Its own publications in the window, newest first, through the bus's sent index.""" return tuple( - message for message in self.source.comms.bus.sent( + message for record in self.source.comms.bus.bus.sent( self.unit.require_recorded().lookup, limit=500) - if message.timestamp >= self.since + for message in record.deliveries() if message.timestamp >= self.since ) @cached_property diff --git a/src/agent_comms/standing_rules.py b/src/agent_comms/standing_rules.py index 18b8240954..a5de04d8e4 100644 --- a/src/agent_comms/standing_rules.py +++ b/src/agent_comms/standing_rules.py @@ -34,7 +34,7 @@ from agent_comms.owner_requests import OwnerRequestId, RequestDirective from agent_comms.recorded_time import EventTime, RecordedEventTime from agent_comms.repository_name import RepositoryName -from agent_comms.repo import GitCommandError, Repo +from agent_comms.repo import GitCommandError from agent_comms.service_passes import ServicePassStore from agent_comms.storage_rules import MemoryUse, StorageRule, UnitStorage from agent_comms.child_process import ProcessIdentity, ResidentMemory @@ -410,8 +410,31 @@ 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 + + def parallel_to(self, other: HolderBranch) -> bool: + """Different holders writing different branches of one repository with shared files.""" + left, right = self.difference, other.difference + return (self.holder != other.holder and left.repository == right.repository + and left.branch != right.branch and bool(left.files & right.files)) + + +@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 @@ -428,12 +451,6 @@ class HeldBranch: def problem(self) -> str: return self.todo.id - def parallel_to(self, other: HeldBranch) -> bool: - """Different holders writing different branches of one repository with shared files.""" - left, right = self.difference, other.difference - return (self.holder != other.holder and left.repository == right.repository - and left.branch != right.branch and bool(left.files & right.files)) - class HolderLine(StandingRule, declared_name=None): """A rule about one holder's order: a recorded action reaching the holder disposes its line.""" @@ -460,6 +477,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 @@ -470,10 +488,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, @@ -576,6 +602,13 @@ class ConflictingBranches(HeldBranchRule): repository: RepositoryName paths: tuple[str, ...] = field(compare=False) + @classmethod + def prepare_captured(cls, prepared: tuple[PreparedBranch, ...], pulls: ProgressRead) -> None: + for first, second in combinations(prepared, 2): + if first.parallel_to(second): + pulls.branch_conflicts(first.repository, first.difference.head, + second.difference.head) + @classmethod def derive(cls, branches: tuple[HeldBranch, ...]) -> tuple[Self, ...]: return tuple( @@ -584,8 +617,8 @@ def derive(cls, branches: tuple[HeldBranch, ...]) -> tuple[Self, ...]: other_branch=second.difference.branch, repository=first.difference.repository, paths=conflicts) for first, second in combinations(branches, 2) - if first.parallel_to(second) and (conflicts := Repo.merge_conflicts( - first.repository.path, first.difference.head, second.difference.head)) + if first.parallel_to(second) and (conflicts := first.source.pulls.require_branch_conflicts( + first.difference.head, second.difference.head)) ) def line(self) -> str: 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.""" From b1b2ce431e2bd8754e82bce88b116198e4541c6b Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Thu, 29 Oct 2026 19:06:23 -0400 Subject: [PATCH 3/3] Select conflict candidates by shared file, not a four-term pair test HolderBranch.parallel_to decided a pair by a hand-written chain over holder, repository, branch and files. The only fact that selects a candidate is a common (repository, file); HolderBranch.touching indexes branches by it and yields each pair once. merge-tree answers the rest: a pair on one branch merges cleanly, and two orders of one holder that do conflict are still a conflict to resolve. unit_talk.py formatted. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01RnH6dR85kyFKcc2tQAJLhK --- src/agent_comms/commander/unit_talk.py | 91 +++++++++++++++++++------- src/agent_comms/standing_rules.py | 30 +++++---- 2 files changed, 84 insertions(+), 37 deletions(-) diff --git a/src/agent_comms/commander/unit_talk.py b/src/agent_comms/commander/unit_talk.py index 2cb7553284..a2aee0ef3b 100644 --- a/src/agent_comms/commander/unit_talk.py +++ b/src/agent_comms/commander/unit_talk.py @@ -49,16 +49,25 @@ def since(self) -> float: 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 + 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}) + return len( + { + point + for origin in self.source.sources + for point in origin.commits() + if point.at >= self.since + } + ) @property def characters(self) -> int: @@ -69,11 +78,16 @@ 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, ...]: + 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)) + return tuple( + rule + for member in cls.members_with(cls) + for talk in talks + for rule in member.derive(talk) + ) @classmethod @abstractmethod @@ -92,16 +106,32 @@ class BlockedTalk(UnitTalkRule, DatedStandingRule): 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", + "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): + 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())) + 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") + 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) @@ -141,7 +180,9 @@ def derive(cls, talk: UnitTalk) -> Iterator[Self]: 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") + 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/standing_rules.py b/src/agent_comms/standing_rules.py index a5de04d8e4..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 @@ -416,11 +416,19 @@ class HolderBranch: holder: ThreadIncarnation difference: BranchDifference - def parallel_to(self, other: HolderBranch) -> bool: - """Different holders writing different branches of one repository with shared files.""" - left, right = self.difference, other.difference - return (self.holder != other.holder and left.repository == right.repository - and left.branch != right.branch and bool(left.files & right.files)) + @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) @@ -604,10 +612,8 @@ class ConflictingBranches(HeldBranchRule): @classmethod def prepare_captured(cls, prepared: tuple[PreparedBranch, ...], pulls: ProgressRead) -> None: - for first, second in combinations(prepared, 2): - if first.parallel_to(second): - pulls.branch_conflicts(first.repository, first.difference.head, - second.difference.head) + 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, ...]: @@ -616,8 +622,8 @@ def derive(cls, branches: tuple[HeldBranch, ...]) -> tuple[Self, ...]: other_problem=second.problem, other_holder=second.holder, other_branch=second.difference.branch, repository=first.difference.repository, paths=conflicts) - for first, second in combinations(branches, 2) - if first.parallel_to(second) and (conflicts := first.source.pulls.require_branch_conflicts( + for first, second in HolderBranch.touching(branches) + if (conflicts := first.source.pulls.require_branch_conflicts( first.difference.head, second.difference.head)) )