P717 stalled orders strike and release themselves

src/agent_comms/commander/order.py

@@ -297,7 +297,8 @@ def send(unit, due, body: str, clean=False, mechanical=False, reassign=None, new
             problem=new[0], repo=Repo.of(Path(check.directory)).github, text=new[1], **common)))
     for pid in [] if new else group:
         call(CommsOrderTool(owner=unit, order=OrderGoalAction(
-            problem=pid, expected_revision=COMMS.goals.todos.get(pid).revision, traced=trace, **common)))
+            problem=pid, expected_revision=COMMS.goals.todos.get(pid).revision,
+            assigner=COMMS.registry.require(ME).incarnation, traced=trace, **common)))
     message += "\n" + "\n".join(order.pull_request_instruction(pid) for pid in group)
     sent = post(VerdictAudience((unit,)), message)
     # The message is each order's recorded link; a verdict citing it lands on that order.
@@ -395,8 +396,8 @@ def board():
              "GENERATED by agent-comms-commander order from Core's Todo store. Do not edit by hand:",
              '`order problem` adds a row, `order send ... Problem: P<n>` orders it, `order close` or `order withdraw` removes it.', '',
              '| Problem | Owner file | Owner | Order | Due | Progress | Delivered '
-             '| Owner last acted |',
-             '|---|---|---|---|---|---|---|---|']
+             '| Strikes | Owner last acted |',
+             '|---|---|---|---|---|---|---|---|---|']
     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):
@@ -405,7 +406,7 @@ def board():
         lines.append(f'| {t.id}: {t.text} | {t.order.owner.to_wire()} | {owner} | {t.order.id} | '
                      f'{hhmm(t.deadline)} | {read[t.id].line if t.id in read else "-"} | '
                      f'{"no" if t.pending else "yes, install pending"} | '
-                     f'{activity[unit].clock() if unit else "-"} |')
+                     f'{t.strike_count} | {activity[unit].clock() if unit else "-"} |')
     history = COMMS.goals.todos.history_for_commander(COMMS.registry.require(ME).incarnation)
     lines += ['', '## Withdrawn orders', '',
               *(f'- {line}' for t in history for line in t.order.withdrawals(t))]
@@ -590,20 +591,18 @@ def replied(latest: Message | None) -> str:
 
 def deadline_lines(pending: list[Todo], read: dict[str, OrderProgress], since: float = 0.0) -> list[str]:
     """Past-due unfinished orders alarm; installed merges have already closed.
-    Stall crossings remain separate from deadline alarms."""
+    Core records a stalled order's strike and release; this view reports deadlines."""
     now = time.time()
-    late, stalled = [], []
+    late = []
     for t in pending:
         ordered, work = t.order.at.recorded_at('An order'), read[t.id]
-        if work.stalled_between(ordered, since, now):
-            stalled.append(f'STALL {t.id} {t.holder}: {work.line} -> push it now or reassign')
         if not t.overdue(now):
             continue
         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)} -> push or merge unfinished work now'
-    return [alarm] * bool(late) + stalled
+    return [alarm] * bool(late)
 
 
 def install_lines(held: tuple[Todo, ...], read: dict[str, OrderProgress], now: float) -> list[str]:
@@ -634,12 +633,6 @@ def overdue():
             for line in lines:
                 print(line, flush=True)
             installs = lines
-        # Units leave finished PRs open and wander off. An order's open PR that nobody has touched for
-        # 5 minutes is a stall, whatever the unit says it is doing; until its deadline, OVERDUE owns the alarm.
-        idle_prs = [f'{t.holder} {t.repo.split("/")[1]} {read[t.id].line} ({t.id})'
-                    for t in pending if t.overdue(now) and read[t.id].untouched(now)]
-        if idle_prs:
-            print('STALL PR open and untouched 5+ min: ' + ', '.join(idle_prs) + ' -> order the merge or the fix now', flush=True)
         for _ in HuntDue.broken(COMMS, ME, since):
             print(HuntVerb.due(next_hunt()), flush=True)
         since = now

src/agent_comms/commander/standing.py

@@ -2,14 +2,18 @@
 
 from __future__ import annotations
 
+import asyncio
+import logging
 import subprocess
 import time
-from collections.abc import Iterator
+from collections.abc import Callable, Iterable, Iterator
+from contextlib import AbstractContextManager
 from dataclasses import dataclass
 from pathlib import Path
 from typing import ClassVar
 
 from ..comms import Comms
+from ..coordinator import Coordination
 from ..goal_actions import OrderGoalAction
 from ..goals import Goal
 from ..standing_rules import (
@@ -18,14 +22,15 @@
     StandingRule,
     UndatedStandingRule,
 )
+from ..thread_identity import StallWatcherIncarnation, ThreadIncarnation
 from ..threads import Thread
-from ..todos import FlatTrend, OrderProgress, PullRequests, StrikeTodo, Todo
+from ..todos import (
+    FlatTrend, OrderProgress, PullRequests, Todo,
+)
+from ..tools import CommsUnjudgedTool
 from ..turn_digests import TurnDigestRead, TurnsSince
-from . import judge
 from .config import CommanderConfig, Repo
 
-CONFIG = CommanderConfig.installed()
-
 
 @dataclass(frozen=True)
 class Detached(UndatedStandingRule):
@@ -56,7 +61,8 @@ class RequestStalled(UndatedStandingRule):
     def violations(cls, comms: Comms, commander: str) -> tuple[RequestStalled, ...]:
         thread = comms.registry.require(commander)
         return tuple(cls(request) for goal in thread.goal.active()
-                     for request, served in goal.order.open_requests(goal, comms.goals, thread.incarnation)
+                     for request, served in goal.order.open_requests(
+                         goal, comms.goals, thread.incarnation)
                      if not served)
 
     def line(self) -> str:
@@ -77,21 +83,22 @@ class WheelSpin(StandingRule, declared_name=None):
     def idle(cls, comms: Comms, commander: str) -> list[Thread]:
         """Units with a recent turn in their own transcript and no order, parked threads aside."""
         held = {todo.holder for todo in cls.todos(comms, commander)}
-        excluded = held | CONFIG.parked | {commander, ""}
+        excluded = held | CommanderConfig.installed().parked | {commander, ""}
         bound = TurnsSince(time.time() - cls.recent)
         return sorted((thread for thread in comms.registry.all_threads().values()
                        if thread.name not in excluded
                        and TurnDigestRead(comms.root, thread, bound).read().turns),
                       key=lambda thread: thread.name)
 
-    def assign(self, comms: Comms, commander: str, unit: Thread) -> None:
+    def assign(self, comms: Comms, commander: str, unit: Thread,
+               assigner: ThreadIncarnation) -> None:
         """The order sent again to ``unit`` through its Goal, with the order's own text."""
-        todo = comms.goals.todos.get(self.todo.id)
+        todo = self.todo
         order = todo.order.reissued(time.time())
         comms.goals.update_goal(unit.name, OrderGoalAction(
             problem=todo.id, expected_revision=todo.revision, order=order,
             request=todo.goal.request(), deadline=todo.deadline, commander=commander,
-            traced=todo.order.owner))
+            assigner=assigner, traced=todo.order.owner))
         comms.messaging.send(commander, unit.name, f"{self.line()}\n{order.instructions(todo)}")
 
 
@@ -122,15 +129,8 @@ def line(self) -> str:
                 " to an idle unit")
 
     def enforce(self, comms: Comms, commander: str) -> None:
-        todo = comms.goals.todos.get(self.todo.id)
-        held = todo.held()
-        if not todo.struck_on_order(held.owner):
-            comms.goals.change_problem(todo.id, StrikeTodo(
-                todo.revision, comms.registry.require(commander).incarnation, held,
-                f"{held.owner} kept working on {todo.id} while the distance to done did not fall"
-                " across two readings"), commander)
         for unit in self.idle(comms, commander)[:1]:
-            self.assign(comms, commander, unit)
+            self.assign(comms, commander, unit, StallWatcherIncarnation.watcher())
 
 
 @dataclass(frozen=True)
@@ -141,17 +141,57 @@ class UnownedWheelSpin(WheelSpin, UndatedStandingRule):
 
     @classmethod
     def violations(cls, comms: Comms, commander: str) -> tuple[UnownedWheelSpin, ...]:
-        rows = (todo for todo in comms.goals.todos.for_commander(
-            comms.registry.require(commander).incarnation) if todo.pending and not todo.holder)
-        idle = cls.idle(comms, commander)
-        return tuple(cls(todo, unit.name) for unit, todo in zip(idle, rows, strict=False))
+        return cls.paired(comms, commander, (todo for todo in
+            comms.goals.todos.for_commander(comms.registry.require(commander).incarnation)
+            if not todo.holder))
+
+    @classmethod
+    def paired(cls, comms: Comms, commander: str,
+               todos: Iterable[Todo]) -> tuple[UnownedWheelSpin, ...]:
+        """Each pending row with the first idle unit not struck on it."""
+        idle, found = cls.idle(comms, commander), []
+        for todo in (todo for todo in todos if todo.pending):
+            for unit in idle:
+                if not todo.live_strikes(unit.name):
+                    idle.remove(unit)
+                    found.append(cls(todo, unit.name))
+                    break
+        return tuple(found)
 
     def line(self) -> str:
         return (f"WHEEL SPIN: {self.unit} holds no order while {self.todo.id} is unowned"
                 f" -> Core gives it to {self.unit}")
 
     def enforce(self, comms: Comms, commander: str) -> None:
-        self.assign(comms, commander, comms.registry.require(self.unit))
+        self.assign(comms, commander, comms.registry.require(self.unit),
+                    comms.registry.require(commander).incarnation)
+
+    @classmethod
+    def after_stall(cls, comms: Comms) -> None:
+        """Assign first: refused or unavailable replacements leave the old holder untouched."""
+        stalled = comms.goals.todos.stalled(comms.registry.storage_rule)
+        for commander in {todo.creator_identity.name for todo in stalled}:
+            mine = (todo for todo in stalled if todo.creator_identity.name == commander)
+            for rule in cls.paired(comms, commander, mine):
+                rule.assign(comms, commander, comms.registry.require(rule.unit),
+                            StallWatcherIncarnation.watcher())
+
+    @classmethod
+    def watch_pass(cls, opened_wire: Callable[[], AbstractContextManager[Comms]]) -> None:
+        """The worker borrows the root for this pass, never while the timer is idle."""
+        with opened_wire() as comms:
+            cls.after_stall(comms)
+
+    @classmethod
+    async def watch(cls, opened_wire: Callable[[], AbstractContextManager[Comms]]) -> None:
+        """Run a pass each minute; failures go to the existing service log."""
+        while True:
+            (outcome,) = await asyncio.gather(
+                Coordination.run_worker(lambda: cls.watch_pass(opened_wire)),
+                return_exceptions=True)
+            if isinstance(outcome, Exception):
+                logging.error("Stall watcher pass failed", exc_info=outcome)
+            await asyncio.sleep(60)
 
 
 @dataclass(frozen=True)
@@ -161,7 +201,7 @@ class Unjudged(ReasonedStandingRule, UndatedStandingRule):
 
     @classmethod
     def reason(cls, comms: Comms, commander: str) -> str | None:
-        total = len(judge.Edit.unjudged(comms, commander))
+        total = CommsUnjudgedTool.count(comms, commander)
         return str(total) if total > cls.limit else None
 
 
@@ -224,7 +264,8 @@ def merges(cls, repo: Repo) -> Iterator[MergeGrowth]:
 
     @classmethod
     def violations(cls, comms: Comms, commander: str) -> tuple[MergeGrowth, ...]:
-        return tuple(growth for repo in CONFIG.repos for growth in cls.merges(repo) if growth.over)
+        return tuple(growth for repo in CommanderConfig.installed().repos
+                     for growth in cls.merges(repo) if growth.over)
 
     @property
     def over(self) -> bool:

src/agent_comms/connector.py

@@ -27,8 +27,8 @@
 import time
 import urllib.request
 from abc import abstractmethod
-from collections.abc import Iterator, Mapping, Sequence
-from contextlib import contextmanager
+from collections.abc import AsyncIterator, Iterator, Mapping, Sequence
+from contextlib import contextmanager, suppress
 from dataclasses import dataclass, field, fields, replace
 from pathlib import Path
 from typing import Annotated, ClassVar, Literal, Self, get_args
@@ -49,6 +49,7 @@
     RegisteredClient, digest,
 )
 from .messages import Message
+from .commander.standing import UnownedWheelSpin
 from .todos import DirectiveText
 from .tool_caller import ThreadCaller
 from .tools import CommsChannelsTool, ReadTool, ToolRequest
@@ -359,6 +360,7 @@ def application(self) -> web.Application:
         with self.opened_wire():
             pass
         app = web.Application()
+        app.cleanup_ctx.append(self.watch_orders)
         app.add_routes(
             [
                 web.get("/.well-known/oauth-protected-resource", self.resource_metadata),
@@ -372,6 +374,16 @@ def application(self) -> web.Application:
         )
         return app
 
+    async def watch_orders(self, app: web.Application) -> AsyncIterator[None]:
+        """The running Core service advances stalled orders without a commander call."""
+        task = asyncio.create_task(UnownedWheelSpin.watch(self.opened_wire))
+        try:
+            yield
+        finally:
+            task.cancel()
+            with suppress(asyncio.CancelledError):
+                await task
+
     async def resource_metadata(self, request: web.Request) -> web.Response:
         return web.json_response(
             {

src/agent_comms/declared_family.py

@@ -13,6 +13,13 @@
 )
 
 
+def declared_words(stem: str) -> str:
+    """A class-name stem as its declared name: ``StallWatcher`` -> ``stall_watcher``."""
+    return re.sub(
+        r"([a-z0-9])([A-Z])", r"\1_\2", re.sub(r"([A-Z]+)([A-Z][a-z])", r"\1_\2", stem)
+    ).lower()
+
+
 class _FamilyMeta(AutoRegisterMeta, ABCMeta):
     def __new__(
         mcs,
@@ -48,9 +55,7 @@ def __new__(
             namespace["declared_name"] = None
         elif root is not None:
             stem = name.removesuffix(root._family_affix) if root._family_affix else name
-            derived = re.sub(
-                r"([a-z0-9])([A-Z])", r"\1_\2", re.sub(r"([A-Z]+)([A-Z][a-z])", r"\1_\2", stem)
-            ).lower()
+            derived = declared_words(stem)
             # A slots dataclass recreates its declaration with the derived name in its namespace.
             namespace["declared_name"] = class_options.pop(
                 "declared_name", namespace.get("declared_name", derived)

src/agent_comms/goal_actions.py

@@ -27,17 +27,18 @@
 from .goal_waits import GoalWaits, RecordedGoalWait
 from .goals import (
     Goal,
-    ReportedTurn,
     GoalDeclaration,
     GoalOrder,
     NoGoal,
     ProblemGoalOrder,
+    ReportedTurn,
     RequestDirective,
     RequestGoalOrder,
     UnorderedGoalOrder,
 )
 from .progress_claims import ProgressClaim
 from .recorded_time import EventTime
+from .thread_identity import ThreadIncarnation
 from .threads import Thread
 from .todos import GoalRef, PendingProblemOrder, SendBackVerdict, Todo, TodoOperation
 
@@ -507,6 +508,9 @@ class ProblemOrderGoalAction(GoalAction[Goal], RuntimeInvocable):
     deadline: EventTime
     commander: str
 
+    @abstractmethod
+    def bound_to(self, actor: ThreadIncarnation) -> ProblemOrderGoalAction: ...
+
     @abstractmethod
     def reserve(self, ctx: GoalActionContext, generation: str, goal: GoalRef) -> Todo: ...
 
@@ -542,12 +546,17 @@ def create(self, ctx: GoalActionContext) -> Goal:
 
 @dataclass(frozen=True, kw_only=True)
 class OrderGoalAction(ProblemOrderGoalAction):
+    """The order's commander gives the problem to this thread."""
     expected_revision: int
     traced: FileLine | None = field(default=None, metadata={"wire_omit_default": True})
+    assigner: ThreadIncarnation
+
+    def bound_to(self, actor: ThreadIncarnation) -> OrderGoalAction:
+        return replace(self, commander=actor.name, assigner=actor)
 
     def reserve(self, ctx: GoalActionContext, generation: str, goal: GoalRef) -> Todo:
         return ctx.goals.todos.assign(self.problem, expected_revision=self.expected_revision,
-            owner=ctx.thread, parent=ctx.goals.registry.require(self.commander),
+            owner=ctx.thread, assigner=self.assigner,
             generation=generation, goal=goal, deadline=self.deadline, order=self.order,
             traced=self.traced)
 
@@ -557,6 +566,9 @@ class CreateProblemGoalAction(ProblemOrderGoalAction):
     repo: str
     text: str
 
+    def bound_to(self, actor: ThreadIncarnation) -> CreateProblemGoalAction:
+        return replace(self, commander=actor.name)
+
     def reserve(self, ctx: GoalActionContext, generation: str, goal: GoalRef) -> Todo:
         return ctx.goals.todos.create(self.problem, self.repo, self.text,
             creator=ctx.goals.registry.require(self.commander), owner=ctx.thread,