Flag stalls and ceremony on the board; sequence only branches that conflict

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"(?<!not )(?<!n't )(?<!n’t )\b(" + "|".join(re.escape(word) for word in words) + r")\b",
+        re.IGNORECASE,
+    )
+    """A negated phrase ("not waiting on", "wasn't waiting for") announces the opposite."""
+    located: ClassVar[re.Pattern[str]] = re.compile(r"[\w./-]+\.\w+:\d+")
+    named_owner: ClassVar[re.Pattern[str]] = re.compile(r"@[\w-]+")
+
+    @classmethod
+    def derive(cls, talk: UnitTalk) -> 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"
+        )

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",)),
         )
     )
 

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."""

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."""
 

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."""