Integrate #2014 #2004

src/agent_comms/agent_activity.py

@@ -4,7 +4,7 @@
 
 import logging
 import time
-from collections.abc import Iterable, Mapping
+from collections.abc import Callable, Iterable, Mapping
 from dataclasses import dataclass, replace
 from functools import partial
 from pathlib import Path
@@ -261,32 +261,27 @@ def begin_turn(
             try:
                 phase = PreparingPhase(detail)
                 lease.record.transition(lease, phase)
-                self._record_turn_activity(leased.name, (phase,))
+                self._emit_activity(phase.activity_event(leased.name))
             except BaseException:
                 self.registry.release_turn(lease)
                 raise
             return RegistryOwner(thread=leased, claimed_generation=lease.claimed_generation)
 
-    def _record_turn_activity(self, name: str, phases: Iterable[TurnPhase]) -> None:
-        """Changed phases are real activity events, just like CLI status reports."""
-        for phase in phases:
-            self._emit_activity(phase.activity_event(name))
+    def _turn_phase_effects(
+        self, lease: TurnLeaseFence, phase: TurnPhase,
+        transition: Callable[[TurnLeaseFence, TurnPhase], tuple[TurnState, ...]],
+    ) -> tuple[TurnState, ...]:
+        with _store_lock(self._wire_lock_path, mode=SharedLock()):
+            effects = transition(lease, phase)
+        for state in effects:
+            state.active.phase.phase_effects(self, lease)
+        return effects
 
     def transition_turn(self, lease: TurnLeaseFence, phase: TurnPhase) -> tuple[TurnState, ...]:
-        with _store_lock(self._wire_lock_path, mode=SharedLock()):
-            effects = lease.record.transition(lease, phase)
-            self._record_turn_activity(
-                lease.identity.incarnation.name, (state.active.phase for state in effects)
-            )
-            return effects
+        return self._turn_phase_effects(lease, phase, lease.record.transition)
 
     def observe_native_phase(self, lease: TurnLeaseFence, phase: TurnPhase) -> tuple[TurnState, ...]:
-        with _store_lock(self._wire_lock_path, mode=SharedLock()):
-            effects = lease.record.observe_native_phase(lease, phase)
-            self._record_turn_activity(
-                lease.identity.incarnation.name, (state.active.phase for state in effects)
-            )
-            return effects
+        return self._turn_phase_effects(lease, phase, lease.record.observe_native_phase)
 
     def finish_turn(self, lease: TurnLeaseFence) -> TurnCompletion:
         """Persist this lease's terminal identity before publishing idle activity."""

src/agent_comms/commander/config.py

@@ -17,6 +17,7 @@
 from ..errors import RelationViolationError
 from ..native_arguments import NativeArguments
 from ..threads import Thread
+from ..todos import Todo
 
 
 class CommanderRole:
@@ -40,6 +41,15 @@ class Repo(BaseModel):
     github: str
     source: str = 'src'
 
+    def contains(self, sha: str) -> bool:
+        """Only a successful ancestry proof in this installed checkout permits closure."""
+        result = subprocess.run(
+            ('git', '-C', str(self.checkout), 'merge-base', '--is-ancestor', sha, 'HEAD'),
+            capture_output=True, timeout=60)
+        if result.returncode not in (0, 1):
+            result.check_returncode()
+        return result.returncode == 0
+
     @classmethod
     def of(cls, checkout: Path) -> Repo:
         """Its GitHub name comes from the checkout's own origin remote."""
@@ -109,6 +119,12 @@ class Commander:
     def of(cls, config: CommanderConfig) -> Commander:
         return cls(config.commander, Comms(config.root))
 
+    def orders(self, repos: tuple[Repo, ...]) -> tuple[Todo, ...]:
+        actor = self.comms.registry.require(self.name).incarnation
+        store = self.comms.goals.todos
+        store.reconcile_merges(actor, repos, self.comms.registry.storage_rule)
+        return store.for_commander(actor)
+
     def units(self) -> frozenset[str]:
         """The threads this commander gives orders to; the owner, claude-ai and peer commanders
         are not among them."""

src/agent_comms/commander/order.py

@@ -397,7 +397,7 @@ def board():
              '| Problem | Owner file | Owner | Order | Due | Progress | Delivered '
              '| Owner last acted |',
              '|---|---|---|---|---|---|---|---|']
-    listed, read = todos(), progress()
+    listed, read = COMMANDER.orders(CONFIG.repos), progress()
     activity = {unit: last_activity(unit) for unit in {t.holder for t in listed if t.holder}}
     for t in sorted(listed, key=lambda t: int(t.id[1:]) if t.id[1:].isdigit() else 0):
         unit, last = t.holder, t.released[-1:]
@@ -589,7 +589,7 @@ def replied(latest: Message | None) -> str:
 
 
 def deadline_lines(pending: list[Todo], read: dict[str, OrderProgress], since: float = 0.0) -> list[str]:
-    """Every past-due unrecorded order alarms with its state until recorded.
+    """Past-due unfinished orders alarm; installed merges have already closed.
     Stall crossings remain separate from deadline alarms."""
     now = time.time()
     late, stalled = [], []
@@ -602,8 +602,7 @@ def deadline_lines(pending: list[Todo], read: dict[str, OrderProgress], since: f
         answer = reply(t.holder, ordered)
         due = f'due {hhmm(t.deadline)}, order {t.order.id}'
         late.append(f'{t.holder} ({due}, {work.line}{replied(answer)})')
-    alarm = (f'OVERDUE: {", ".join(late)} -> record merged work; '
-             'push or merge unfinished work now')
+    alarm = f'OVERDUE: {", ".join(late)} -> push or merge unfinished work now'
     return [alarm] * bool(late) + stalled
 
 
@@ -618,7 +617,7 @@ def overdue():
     installs: list[str] = []
     while True:
         now = time.time()
-        held = held_orders()
+        held = tuple(todo for todo in COMMANDER.orders(CONFIG.repos) if todo.assignment)
         activity = {unit: last_activity(unit) for unit in {t.assignment.owner for t in held}}
         if missing := first_calls(held, activity):
             print('NO FIRST CALL: ' + ', '.join(missing) + ' -> stuck holding an order; push it now or reassign',

src/agent_comms/pi_payloads.py

@@ -18,6 +18,7 @@
 from .declared_family import DeclaredFamily, UnknownFamilyMember
 from .field_codec import FieldCodec, Id128Text, MalformedRecord, TextRepresentation
 from .native_file_artifact import NativeFileArtifact
+from .turn_failure import ProviderTransportFailed
 from .native_session_reopen import NativeSessionIdentity
 from .pi_counts import NonnegativePiCount, PiCount, UnreportedCount
 from .pi_vocabulary import (
@@ -33,6 +34,7 @@
 
 if TYPE_CHECKING:
     from .agent_events import ProviderUsage
+    from .turn_phase import TurnPhase
     from .turn_usage import UsageAccount
 
 
@@ -504,6 +506,9 @@ def wire_member(cls, value):
     def refine_model_request_failure(self, failure):
         return failure
 
+    def turn_phase(self, phase: TurnPhase) -> TurnPhase:
+        return phase
+
     @property
     def description(self) -> str:
         return ""
@@ -516,9 +521,11 @@ class ProviderTransportFailureDiagnostic(PiDiagnostic):
     details: ProviderTransportDetails
     timestamp: int | None = None
 
-    def refine_model_request_failure(self, failure):
-        from .turn_failure import ProviderTransportFailed
+    def turn_phase(self, phase: TurnPhase) -> TurnPhase:
+        from .turn_phase import FailedPhase
+        return FailedPhase(error=self.error)
 
+    def refine_model_request_failure(self, failure):
         return ProviderTransportFailed(failure.text)
 
     @property

src/agent_comms/pi_vocabulary.py

@@ -152,6 +152,8 @@ async def apply(cls, session, message):
             return
         from .acp_failure import NativeMessageFailureSource
 
+        for diagnostic in message.diagnostics:
+            session.watchdog.phase = diagnostic.turn_phase(session.watchdog.phase)
         error = session.output.error(
             (message.error_message or "").strip() or f"Model request {cls.declared_name}",
             message.diagnostics, source=NativeMessageFailureSource(message),

src/agent_comms/standing_rules.py

@@ -196,7 +196,7 @@ def violations(cls, comms: Comms, commander: str) -> tuple[Self, ...]:
     def line(self) -> str:
         return (
             f"BOARD DISAGREES: {self.problem} order {self.order} reads no commit, "
-            f"but its linked PR merged as {self.merge} -> repair the progress read this wake"
+            f"but its linked PR merged as {self.merge[:9]} -> repair the progress read this wake"
         )
 
 

src/agent_comms/todos.py

@@ -45,6 +45,7 @@
 from .plan_phases import OrderPhase, UnplannedPhase
 
 if TYPE_CHECKING:
+    from .commander.config import Repo
     from .comms import Comms
     from .goal_management import Goals
 
@@ -590,18 +591,18 @@ def untouched(self, now: float) -> bool:
 
 @dataclass(frozen=True)
 class MergedProgress(FoundProgress):
-    """A merge awaits its order record; its alarm does not itself draw a strike."""
+    """A linked merge closes its order once the installed checkout contains it."""
     precedence: ClassVar[int] = 1
     sha: str
 
     @classmethod
     def found(cls, source: ProgressSource) -> Iterator[Self]:
-        return (cls(pr.merge[:9]) for pr in source.requests("merged")
+        return (cls(pr.merge) for pr in source.requests("merged")
                 if pr.merge and (pr.merged_at or 0) >= source.since)
 
     @property
     def state(self) -> str:
-        return f"merged as {self.sha}, awaiting record"
+        return f"merged as {self.sha[:9]}, waiting on install"
 
     def waiting_on_install(self, todo: Todo) -> tuple[Todo, ...]:
         return (todo,)
@@ -1238,6 +1239,9 @@ def changed(name: str) -> bool:
     def copy_as[Kind: ProblemOrder](self, kind: type[Kind], **evidence) -> Kind:
         return kind(**{item.name: getattr(self, item.name) for item in fields(self)}, **evidence)
 
+    def merged(self, sha: str) -> ProblemOrder:
+        return MergedProblemOrder(original=self, sha=sha)
+
     @property
     def pull_request_reference(self) -> str:
         return f"{self.pull_request_label} {self.id}"
@@ -1310,6 +1314,21 @@ def waiting_on_install(self, todo: Todo, read: dict[str, OrderProgress]) -> tupl
         return ()
 
 
+@dataclass(frozen=True, kw_only=True)
+class MergedProblemOrder(ProblemOrder):
+    """An installed merge, retaining its original order without inventing a check or app proof."""
+    original: CheckedProblemOrder
+    sha: str
+    line: ClassVar[str] = "Merged and installed"
+
+    @property
+    def id(self) -> str:
+        return self.original.id
+
+    def require_close(self) -> None:
+        pass
+
+
 @dataclass(frozen=True, kw_only=True)
 class PendingProblemOrder(CheckedProblemOrder):
     def reissued(self, at: float) -> PendingProblemOrder:
@@ -1626,7 +1645,13 @@ def change(cls, current, command):
     @classmethod
     def decide(cls, current, command):
         current.require_commander(command.actor)
-        return replace(current, state=cls, assignment=None, revision=current.revision + 1)
+        return cls.closed(current, current.order)
+
+    @classmethod
+    def closed(cls, current: Todo, order: ProblemOrder) -> Todo:
+        order.require_close()
+        return replace(current, state=cls, order=order, assignment=None,
+                       revision=current.revision + 1)
 
 
 @dataclass(frozen=True)
@@ -2297,6 +2322,26 @@ def for_goal(self, owner: ThreadIncarnation, goal_id: str) -> tuple[Todo, ...]:
             "json_extract(assignment,'$.owner')=? AND json_extract(assignment,'$.owner_created')=? "
             "AND json_extract(assignment,'$.generation')=?", (owner.name, owner.created_at, goal_id))
 
+    def reconcile_merges(self, commander: ThreadIncarnation, repos: Sequence[Repo],
+                         storage: StorageRule) -> None:
+        """Read before locking; close only orders whose observed revision still holds."""
+        checked = tuple((todo, order) for todo in self.held_for_commander(commander)
+                        if isinstance(order := todo.order, CheckedProblemOrder))
+        pulls = PullRequests(tuple(todo for todo, _ in checked))
+        completed = tuple(DoneTodoState.closed(todo, order.merged(merged.sha))
+                          for todo, order in checked
+                          for repo in repos if repo.github.lower() == todo.repo
+                          for merged in MergedProgress.found(
+                              ProgressSource.of(todo, storage, pulls))
+                          if repo.contains(merged.sha))
+        if not completed:
+            return
+        with self.writing() as db:
+            for result in completed:
+                current = self._current(db, result.id)
+                if current.revision + 1 == result.revision:
+                    result.upsert(db)
+
     def board_disagreements(
         self, commander: ThreadIncarnation, storage: StorageRule,
     ) -> Iterator[tuple[Todo, MergedProgress]]:

src/agent_comms/turn_phase.py

@@ -7,8 +7,11 @@
 
 from .declared_family import DeclaredFamily
 from .activity import Activity, ActivityState, ObservedActivity
+from .pi_payloads import ProviderDiagnosticError
 if TYPE_CHECKING:
     from . import pi_events as pi
+    from .agent_activity import AgentActivity
+    from .turn_lease import TurnLeaseFence
 
 
 class StallExempt:
@@ -121,6 +124,11 @@ def response_output(self) -> TurnPhase:
     def stalled(self, accepted: bool) -> TurnPhase:
         return self if accepted else PromptAcceptancePhase()
 
+    def phase_effects(self, agents: AgentActivity, lease: TurnLeaseFence) -> None:
+        agents.set_activity(
+            lease.identity.incarnation.name, self.activity_state, self.activity_detail,
+        )
+
 
 class Excursion(TurnPhase, declared_name=None):
     pauses_input_clock = True
@@ -146,6 +154,14 @@ class IdlePhase(ActivityWord, TurnPhase):
     activity_state = ActivityState.IDLE
     label = "Ready"
 
+    def phase_effects(self, agents: AgentActivity, lease: TurnLeaseFence) -> None:
+        agents.finish_turn(lease)
+
+
+@dataclass(frozen=True, kw_only=True)
+class FailedPhase(IdlePhase):
+    error: ProviderDiagnosticError
+
 
 class CancellingPhase(TurnPhase):
     accepts_followup = False