From be7e29a8206b2fcd92a1f4f4b74b4c937779e95e Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Mon, 19 Oct 2026 14:48:21 -0400 Subject: [PATCH 01/18] P718 make stalled orders strike and release themselves Stalling owns the 600-second clock and the atomic strike/release transition. Core connector observes activity without running each order Check. Delete commander STALL prompts and automatic fallback assignment. Released rows expose their recorded strikes. Src-Budget: 160 --- src/agent_comms/commander/order.py | 20 ++---- src/agent_comms/commander/standing.py | 25 ++------ src/agent_comms/connector.py | 15 ++++- src/agent_comms/todos.py | 92 +++++++++++++++++++++++---- 4 files changed, 104 insertions(+), 48 deletions(-) diff --git a/src/agent_comms/commander/order.py b/src/agent_comms/commander/order.py index 92f37f8993..dada7f9085 100644 --- a/src/agent_comms/commander/order.py +++ b/src/agent_comms/commander/order.py @@ -391,8 +391,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` 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 = todos(), 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): @@ -401,7 +401,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))] @@ -586,13 +586,11 @@ 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. - Stall crossings remain separate from deadline alarms.""" + Core records a stalled order's strike and release; this view only 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) @@ -600,7 +598,7 @@ def deadline_lines(pending: list[Todo], read: dict[str, OrderProgress], since: f late.append(f'{t.holder} ({due}, {work.line}{replied(answer)})') alarm = (f'OVERDUE: {", ".join(late)} -> record merged work; ' '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]: @@ -631,12 +629,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 diff --git a/src/agent_comms/commander/standing.py b/src/agent_comms/commander/standing.py index b8408a26b7..7a1b3619f8 100644 --- a/src/agent_comms/commander/standing.py +++ b/src/agent_comms/commander/standing.py @@ -10,7 +10,6 @@ from typing import ClassVar from ..comms import Comms -from ..goal_actions import OrderGoalAction from ..standing_rules import (DatedStandingRule, ReasonedStandingRule, StandingRule, UndatedStandingRule) from ..goals import Goal @@ -62,8 +61,7 @@ def line(self) -> str: @dataclass(frozen=True) class WheelSpin(StandingRule, declared_name=None): - """An order is not getting closer to done, or a unit sits with no order; Core gives the - order to a unit that will move it, with no deadline and no commander turn involved.""" + """Progress problems read from Core; the commander chooses the next assignee.""" todo: Todo @@ -80,16 +78,6 @@ def idle(cls, comms: Comms, commander: str) -> list[Thread]: and TurnDigestRead(comms.root, thread, bound).native()), key=lambda thread: thread.name) - def assign(self, comms: Comms, commander: str, unit: Thread) -> None: - """The order sent again to ``unit`` through its Goal, with the order's own text.""" - todo = comms.goals.todos.get(self.todo.id) - 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)) - comms.messaging.send(commander, unit.name, f"{self.line()}\n{order.instructions(todo)}") - @dataclass(frozen=True) class FlatWheelSpin(WheelSpin, DatedStandingRule): @@ -114,8 +102,8 @@ def began(self) -> float: def line(self) -> str: return (f"WHEEL SPIN: {self.todo.holder} kept working on {self.todo.id} while its distance" - " to done did not fall across two readings -> Core strikes it and gives the order" - " to an idle unit") + " to done did not fall across two readings -> Core strikes it; the commander" + " chooses the next unit") def enforce(self, comms: Comms, commander: str) -> None: todo = comms.goals.todos.get(self.todo.id) @@ -125,8 +113,6 @@ def enforce(self, comms: Comms, commander: str) -> None: 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) @dataclass(frozen=True) @@ -144,10 +130,7 @@ def violations(cls, comms: Comms, commander: str) -> tuple[UnownedWheelSpin, ... 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)) + f" -> the commander must pick its next owner") @dataclass(frozen=True) diff --git a/src/agent_comms/connector.py b/src/agent_comms/connector.py index a584d8825c..d0c023ff12 100644 --- a/src/agent_comms/connector.py +++ b/src/agent_comms/connector.py @@ -28,7 +28,7 @@ import urllib.request from abc import abstractmethod from collections.abc import Iterator, Mapping, Sequence -from contextlib import contextmanager +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 @@ -366,6 +366,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), @@ -379,6 +380,18 @@ def application(self) -> web.Application: ) return app + async def watch_orders(self, app: web.Application): + """The running Core service advances stalled orders without a commander call.""" + with self.opened_wire() as comms: + task = asyncio.create_task( + comms.goals.todos.watch_stalled(comms.registry.storage_rule)) + 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( { diff --git a/src/agent_comms/todos.py b/src/agent_comms/todos.py index 21893e014e..62df9d9986 100644 --- a/src/agent_comms/todos.py +++ b/src/agent_comms/todos.py @@ -353,6 +353,14 @@ def git(self, *args: str) -> str: raise ValueError(run.stderr.strip() or f"git {args[0]} failed") return run.stdout + def readings(self) -> DistanceReadings: + """Progress observation does not run the order's Check.""" + return DistanceReadings() + + +class MeasuredProgressSource(ProgressSource): + """The board's distance view admits Check execution; the watchdog only reads activity.""" + def readings(self) -> DistanceReadings: """The order's base, then each branch commit since the order; the last two are read.""" base = self.git("merge-base", "HEAD", "origin/HEAD").strip() @@ -384,7 +392,11 @@ def read(cls, todo: Todo, storage: StorageRule, pulls: PullRequests) -> OrderPro """The first found member by its declared precedence; with none found, no commit yet.""" owner = todo.holder assert owner, "Only a held order has progress." - source = ProgressSource(todo, storage.order_checkout(owner, todo.id), pulls) + return cls.from_source(MeasuredProgressSource( + todo, storage.order_checkout(owner, todo.id), pulls)) + + @classmethod + def from_source(cls, source: ProgressSource) -> OrderProgress: members = sorted(cls.members_with(FoundProgress), key=lambda member: member.precedence) try: return next((progress for member in members for progress in member.found(source)), @@ -404,6 +416,10 @@ def distance_readings(self) -> DistanceReadings: """The order's distance readings; progress that read no commit has none.""" return DistanceReadings() + def advance(self, todo: Todo, now: float) -> Todo: + """Progress with no stall consequence leaves its order unchanged.""" + return todo + def stalled(self, ordered: float, now: float) -> bool: return False @@ -436,26 +452,46 @@ def found(cls, source: ProgressSource) -> Iterator[Self]: class Stalling: - """Work that stalls at a time its own class declares.""" + """One inactivity deadline and its strike/release transition.""" + + seconds: ClassVar[int] = 600 + + @abstractmethod + def last_move(self, ordered: float) -> float: ... @abstractmethod - def stall_at(self, ordered: float) -> float: ... + def stall_reason(self) -> str: ... + + def stall_at(self, ordered: float) -> float: + return self.last_move(ordered) + self.seconds def stalled(self, ordered: float, now: float) -> bool: - return now > self.stall_at(ordered) + return now >= self.stall_at(ordered) + + def advance(self, todo: Todo, now: float) -> Todo: + if not self.stalled(todo.order.at.recorded_at("An order"), now): + return todo + owner, actor = todo.held(), todo.creator_identity + reason = StrikeReason(f"{owner.owner} stalled on {todo.id}: {self.stall_reason()}") + at = RecordedEventTime(now) + struck = StrikeTodo(todo.revision, actor, owner, reason.text, at).apply(todo) + return ReleaseTodo(struck.revision, actor, owner, reason, at).apply(struck) def stalled_between(self, ordered: float, since: float, now: float) -> bool: - return since < self.stall_at(ordered) < now + return since < self.stall_at(ordered) <= now @dataclass(frozen=True) class Moving(Stalling): - """Work last moved ``at``; it stalls 15 minutes later with no commit, push or PR change.""" + """Work whose last commit, push or PR change was observed at ``at``.""" at: float = field(kw_only=True) readings: DistanceReadings = field(default=DistanceReadings(), kw_only=True, compare=False) - def stall_at(self, ordered: float) -> float: - return self.at + 900 + def last_move(self, ordered: float) -> float: + return max(ordered, self.at) + + def stall_reason(self) -> str: + return f"no commit, push or PR change for ten minutes after {self.state}" def distance_readings(self) -> DistanceReadings: return self.readings @@ -470,9 +506,11 @@ def missing_merges(self, source: ProgressSource) -> tuple[MergedProgress, ...]: def state(self) -> str: return "no commit since the order" - def stall_at(self, ordered: float) -> float: - """No first commit within 10 minutes of the order.""" - return ordered + 600 + def last_move(self, ordered: float) -> float: + return ordered + + def stall_reason(self) -> str: + return "no commit, push or PR change for ten minutes after the order" @dataclass(frozen=True) @@ -1994,7 +2032,7 @@ def held(self) -> Assignment: @property def strike_count(self) -> int: - return 0 if self.assignment is None else len(self.live_strikes(self.assignment.owner)) + return len(self.live_strikes(self.holder)) def live_strikes(self, unit: str | None = None) -> tuple[StrikeTodo, ...]: """The strikes still counting on this task, against ``unit`` on any order of it.""" @@ -2233,6 +2271,36 @@ 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 advance_stalled(self, storage: StorageRule) -> tuple[Todo, ...]: + """Observe outside SQLite, then atomically strike and release the same assignment.""" + held = tuple(todo for todo in self._open("assignment IS NOT NULL", ()) if todo.pending) + pulls = PullRequests(held) + changed = [] + for todo in held: + observed_at = time.time() + source = ProgressSource(todo, storage.order_checkout(todo.held().owner, todo.id), pulls) + progress = OrderProgress.from_source(source) + try: + with self.writing() as db: + current = self._current(db, todo.id) + current.require_result(todo) + result = progress.advance(current, observed_at) + if result != current: + result.upsert(db) + changed.append(result) + except TodoConflict: + # A concurrent order change owns the row; observe its new assignment next pass. + continue + return tuple(changed) + + async def watch_stalled(self, storage: StorageRule) -> None: + """The backend observes Git/PR changes; Stalling alone owns the deadline.""" + from .coordinator import Coordination + + while True: + await Coordination.run_worker(lambda: self.advance_stalled(storage)) + await asyncio.sleep(60) + def board_disagreements( self, commander: ThreadIncarnation, storage: StorageRule, ) -> Iterator[tuple[Todo, MergedProgress]]: From 2a1af9fb17b18760848280f3386d8dd4ba044eca Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Mon, 19 Oct 2026 14:54:13 -0400 Subject: [PATCH 02/18] P717 derive last activity through the found progress capability Cache Git movement per observation; GitProgress applies it once to found commits, pushes and open PRs. Missing checkout is a declared progress state, not an exception fallback. Git read and decode faults raise; only timeout becomes UnreadProgress. Src-Budget: 160 --- src/agent_comms/todos.py | 84 ++++++++++++++++++++++++++++++++-------- 1 file changed, 67 insertions(+), 17 deletions(-) diff --git a/src/agent_comms/todos.py b/src/agent_comms/todos.py index 62df9d9986..50efbfb41e 100644 --- a/src/agent_comms/todos.py +++ b/src/agent_comms/todos.py @@ -17,6 +17,7 @@ from abc import abstractmethod from collections.abc import Iterator, Sequence from dataclasses import dataclass, field, fields, replace +from functools import cached_property from pathlib import Path from typing import TYPE_CHECKING, ClassVar, Final, Self from uuid import uuid4 @@ -337,15 +338,24 @@ def requests(self, state: str) -> list[PullRequest]: def commits(self, *excluded: str) -> Iterator[tuple[str, float]]: """The checkout's newest commit since the order that no ``excluded`` ref already holds.""" - after = time.strftime('%Y-%m-%dT%H:%M:%SZ', time.gmtime(self.since)) - log = subprocess.run( - ("git", "-C", str(self.checkout), "log", "-1", f"--since={after}", - "--format=%H %ct", "HEAD", "--not", *excluded), - capture_output=True, text=True, timeout=30) - for line in log.stdout.splitlines() if not log.returncode else (): + for line in self.git("log", "-1", f"--since={self.git_since}", + "--format=%H %ct", "HEAD", "--not", *excluded).splitlines(): sha, at = line.split() yield sha, float(at) + @property + def git_since(self) -> str: + return time.strftime('%Y-%m-%dT%H:%M:%SZ', time.gmtime(self.since)) + + @cached_property + def pushes(self) -> tuple[PushedProgress, ...]: + return tuple(PushedProgress.pushes(self)) + + @cached_property + def moved_at(self) -> float: + return max((self.since, *(at for _, at in self.commits()), + *(push.at for push in self.pushes))) + def git(self, *args: str) -> str: run = subprocess.run(("git", "-C", str(self.checkout), *args), capture_output=True, text=True, timeout=60) @@ -364,9 +374,8 @@ class MeasuredProgressSource(ProgressSource): def readings(self) -> DistanceReadings: """The order's base, then each branch commit since the order; the last two are read.""" base = self.git("merge-base", "HEAD", "origin/HEAD").strip() - after = time.strftime('%Y-%m-%dT%H:%M:%SZ', time.gmtime(self.since)) commits = [(sha, float(at)) for sha, at in (line.split() for line in reversed(self.git( - "log", "--first-parent", f"--since={after}", "--format=%H %ct", "HEAD", + "log", "--first-parent", f"--since={self.git_since}", "--format=%H %ct", "HEAD", "--not", base).splitlines()))] points = [(base, self.since), *commits][-2:] return DistanceReadings( @@ -399,7 +408,7 @@ def read(cls, todo: Todo, storage: StorageRule, pulls: PullRequests) -> OrderPro def from_source(cls, source: ProgressSource) -> OrderProgress: members = sorted(cls.members_with(FoundProgress), key=lambda member: member.precedence) try: - return next((progress for member in members for progress in member.found(source)), + return next((progress for member in members for progress in member.observations(source)), NoCommitProgress()) except subprocess.TimeoutExpired as unread: return UnreadProgress(str(unread)) @@ -450,6 +459,10 @@ class FoundProgress(OrderProgress): def found(cls, source: ProgressSource) -> Iterator[Self]: """Each member of this class the source holds.""" + @classmethod + def observations(cls, source: ProgressSource) -> Iterator[Self]: + return cls.found(source) + class Stalling: """One inactivity deadline and its strike/release transition.""" @@ -497,6 +510,20 @@ def distance_readings(self) -> DistanceReadings: return self.readings +@dataclass(frozen=True, kw_only=True) +class MissingCheckoutProgress(Moving, FoundProgress): + precedence: ClassVar[int] = 2 + state: ClassVar[str] = "ordered checkout is missing" + + @classmethod + def found(cls, source: ProgressSource) -> Iterator[Self]: + if not source.checkout.exists(): + yield cls(at=max((source.since, *(pr.updated_at for pr in source.requests("open"))))) + + def stall_reason(self) -> str: + return "the ordered checkout is missing and its PR has not changed for ten minutes" + + @dataclass(frozen=True) class NoCommitProgress(Stalling, OrderProgress): def missing_merges(self, source: ProgressSource) -> tuple[MergedProgress, ...]: @@ -522,9 +549,18 @@ def state(self) -> str: return f"progress unread: {self.reason}" +class GitProgress(Moving, FoundProgress, declared_name=None): + """Found progress whose activity includes the checkout's commits and pushes.""" + + @classmethod + def observations(cls, source: ProgressSource) -> Iterator[Self]: + return (replace(progress, at=max(progress.at, source.moved_at)) + for progress in cls.found(source)) + + @dataclass(frozen=True) -class CommittedProgress(Moving, FoundProgress): - precedence: ClassVar[int] = 3 +class CommittedProgress(GitProgress): + precedence: ClassVar[int] = 4 sha: str @classmethod @@ -540,12 +576,26 @@ def state(self) -> str: @dataclass(frozen=True) class PushedProgress(CommittedProgress): """Pushed past the default branch, with no PR naming the order yet.""" - precedence: ClassVar[int] = 4 + precedence: ClassVar[int] = 5 + + @classmethod + def pushes(cls, source: ProgressSource) -> Iterator[Self]: + """A push-ref update has its own time, not the commit's author time.""" + branch = source.git("branch", "--show-current").strip() + for row in source.git("reflog", "show", "--all", "--date=unix", + f"--since={source.git_since}", + "--format=%gD%x09%H%x09%gs").splitlines(): + selector, sha, subject = row.split("\t", 2) + reference, stamp = selector.rsplit("@{", 1) + if reference == f"refs/remotes/origin/{branch}" and subject == "update by push": + yield cls(sha[:9], at=float(stamp.removesuffix("}"))) @classmethod def found(cls, source: ProgressSource) -> Iterator[Self]: + moves = (*source.commits("origin/HEAD"), + *((push.sha, push.at) for push in source.pushes)) return (cls(sha[:9], at=at, readings=source.readings()) - for sha, at in source.commits("origin/HEAD")) + for sha, at in sorted(moves, key=lambda move: move[1], reverse=True)[:1]) @property def state(self) -> str: @@ -553,15 +603,15 @@ def state(self) -> str: @dataclass(frozen=True) -class OpenPrProgress(Moving, FoundProgress): - precedence: ClassVar[int] = 2 +class OpenPrProgress(GitProgress): + precedence: ClassVar[int] = 3 number: int head: str @classmethod def found(cls, source: ProgressSource) -> Iterator[Self]: - return (cls(pr.number, pr.head[:9], at=pr.updated_at, readings=source.readings()) - for pr in source.requests("open")) + return (cls(pr.number, pr.head[:9], at=pr.updated_at, + readings=source.readings()) for pr in source.requests("open")) @property def state(self) -> str: From 38e0d88ac6bfa6b97c34f952b1b4dcdf6fce16ee Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Mon, 19 Oct 2026 15:17:01 -0400 Subject: [PATCH 03/18] P717 restore idle-unit reassignment, derive stall window once, top-level Coordination import WheelSpin.assign and both enforce calls come back from main; a unit struck on a problem is never paired with it again. Stalling.stall_reason writes the window from Stalling.seconds once; the three ten-minute literals are gone. coordinator imports no todos-dependent module, so the local import hid no cycle and moves to module level. Co-Authored-By: Claude Opus 5.5 --- src/agent_comms/commander/standing.py | 34 ++++++++++++++++++++++----- src/agent_comms/todos.py | 16 ++++--------- 2 files changed, 32 insertions(+), 18 deletions(-) diff --git a/src/agent_comms/commander/standing.py b/src/agent_comms/commander/standing.py index 7a1b3619f8..d66a7fff90 100644 --- a/src/agent_comms/commander/standing.py +++ b/src/agent_comms/commander/standing.py @@ -10,6 +10,7 @@ from typing import ClassVar from ..comms import Comms +from ..goal_actions import OrderGoalAction from ..standing_rules import (DatedStandingRule, ReasonedStandingRule, StandingRule, UndatedStandingRule) from ..goals import Goal @@ -61,7 +62,8 @@ def line(self) -> str: @dataclass(frozen=True) class WheelSpin(StandingRule, declared_name=None): - """Progress problems read from Core; the commander chooses the next assignee.""" + """An order is not getting closer to done, or a unit sits with no order; Core gives the + order to a unit that will move it, with no deadline and no commander turn involved.""" todo: Todo @@ -78,6 +80,16 @@ def idle(cls, comms: Comms, commander: str) -> list[Thread]: and TurnDigestRead(comms.root, thread, bound).native()), key=lambda thread: thread.name) + def assign(self, comms: Comms, commander: str, unit: Thread) -> None: + """The order sent again to ``unit`` through its Goal, with the order's own text.""" + todo = comms.goals.todos.get(self.todo.id) + 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)) + comms.messaging.send(commander, unit.name, f"{self.line()}\n{order.instructions(todo)}") + @dataclass(frozen=True) class FlatWheelSpin(WheelSpin, DatedStandingRule): @@ -102,8 +114,8 @@ def began(self) -> float: def line(self) -> str: return (f"WHEEL SPIN: {self.todo.holder} kept working on {self.todo.id} while its distance" - " to done did not fall across two readings -> Core strikes it; the commander" - " chooses the next unit") + " to done did not fall across two readings -> Core strikes it and gives the order" + " to an idle unit") def enforce(self, comms: Comms, commander: str) -> None: todo = comms.goals.todos.get(self.todo.id) @@ -113,6 +125,8 @@ def enforce(self, comms: Comms, commander: str) -> None: 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) @dataclass(frozen=True) @@ -125,12 +139,20 @@ class UnownedWheelSpin(WheelSpin, UndatedStandingRule): 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)) + idle, found = cls.idle(comms, commander), [] + for todo in rows: # a unit struck on a problem never gets that problem back + unit = next((unit for unit in idle if not todo.live_strikes(unit.name)), None) + if unit is not None: + idle.remove(unit) + found.append(cls(todo, unit.name)) + return tuple(found) def line(self) -> str: return (f"WHEEL SPIN: {self.unit} holds no order while {self.todo.id} is unowned" - f" -> the commander must pick its next owner") + f" -> Core gives it to {self.unit}") + + def enforce(self, comms: Comms, commander: str) -> None: + self.assign(comms, commander, comms.registry.require(self.unit)) @dataclass(frozen=True) diff --git a/src/agent_comms/todos.py b/src/agent_comms/todos.py index 50efbfb41e..bf13d58769 100644 --- a/src/agent_comms/todos.py +++ b/src/agent_comms/todos.py @@ -24,6 +24,7 @@ from .change_report import FileLine from .child_process import BoundedRun, ChildResult, ExitedOutcome +from .coordinator import Coordination from .command import Command from .declared_family import DeclaredFamily from .errors import ToolRefusedError @@ -472,8 +473,9 @@ class Stalling: @abstractmethod def last_move(self, ordered: float) -> float: ... - @abstractmethod - def stall_reason(self) -> str: ... + def stall_reason(self) -> str: + """The strike reason, with the window this class declares in ``seconds``.""" + return f"{self.state}; no commit, push or PR change for {self.seconds // 60} minutes" def stall_at(self, ordered: float) -> float: return self.last_move(ordered) + self.seconds @@ -503,9 +505,6 @@ class Moving(Stalling): def last_move(self, ordered: float) -> float: return max(ordered, self.at) - def stall_reason(self) -> str: - return f"no commit, push or PR change for ten minutes after {self.state}" - def distance_readings(self) -> DistanceReadings: return self.readings @@ -520,9 +519,6 @@ def found(cls, source: ProgressSource) -> Iterator[Self]: if not source.checkout.exists(): yield cls(at=max((source.since, *(pr.updated_at for pr in source.requests("open"))))) - def stall_reason(self) -> str: - return "the ordered checkout is missing and its PR has not changed for ten minutes" - @dataclass(frozen=True) class NoCommitProgress(Stalling, OrderProgress): @@ -536,9 +532,6 @@ def state(self) -> str: def last_move(self, ordered: float) -> float: return ordered - def stall_reason(self) -> str: - return "no commit, push or PR change for ten minutes after the order" - @dataclass(frozen=True) class UnreadProgress(OrderProgress): @@ -2345,7 +2338,6 @@ def advance_stalled(self, storage: StorageRule) -> tuple[Todo, ...]: async def watch_stalled(self, storage: StorageRule) -> None: """The backend observes Git/PR changes; Stalling alone owns the deadline.""" - from .coordinator import Coordination while True: await Coordination.run_worker(lambda: self.advance_stalled(storage)) From f4ebfdd9b0810ec5fecd03b8b5c7b853638360ee Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Mon, 19 Oct 2026 15:23:24 -0400 Subject: [PATCH 04/18] P717 stall advance takes the order time from its progress source; typed watcher Co-Authored-By: Claude Opus 5.5 --- src/agent_comms/commander/standing.py | 9 +++++---- src/agent_comms/connector.py | 4 ++-- src/agent_comms/todos.py | 18 +++++++++++------- 3 files changed, 18 insertions(+), 13 deletions(-) diff --git a/src/agent_comms/commander/standing.py b/src/agent_comms/commander/standing.py index d66a7fff90..c17c1a032a 100644 --- a/src/agent_comms/commander/standing.py +++ b/src/agent_comms/commander/standing.py @@ -141,10 +141,11 @@ def violations(cls, comms: Comms, commander: str) -> tuple[UnownedWheelSpin, ... comms.registry.require(commander).incarnation) if todo.pending and not todo.holder) idle, found = cls.idle(comms, commander), [] for todo in rows: # a unit struck on a problem never gets that problem back - unit = next((unit for unit in idle if not todo.live_strikes(unit.name)), None) - if unit is not None: - idle.remove(unit) - found.append(cls(todo, unit.name)) + 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: diff --git a/src/agent_comms/connector.py b/src/agent_comms/connector.py index d0c023ff12..a6c0c43723 100644 --- a/src/agent_comms/connector.py +++ b/src/agent_comms/connector.py @@ -27,7 +27,7 @@ import time import urllib.request from abc import abstractmethod -from collections.abc import Iterator, Mapping, Sequence +from collections.abc import AsyncIterator, Iterator, Mapping, Sequence from contextlib import contextmanager, suppress from dataclasses import dataclass, field, fields, replace from pathlib import Path @@ -380,7 +380,7 @@ def application(self) -> web.Application: ) return app - async def watch_orders(self, app: web.Application): + async def watch_orders(self, app: web.Application) -> AsyncIterator[None]: """The running Core service advances stalled orders without a commander call.""" with self.opened_wire() as comms: task = asyncio.create_task( diff --git a/src/agent_comms/todos.py b/src/agent_comms/todos.py index bf13d58769..705cb6a8a7 100644 --- a/src/agent_comms/todos.py +++ b/src/agent_comms/todos.py @@ -370,7 +370,7 @@ def readings(self) -> DistanceReadings: class MeasuredProgressSource(ProgressSource): - """The board's distance view admits Check execution; the watchdog only reads activity.""" + """The board's distance view runs the Check; the stall watcher only reads activity.""" def readings(self) -> DistanceReadings: """The order's base, then each branch commit since the order; the last two are read.""" @@ -409,8 +409,8 @@ def read(cls, todo: Todo, storage: StorageRule, pulls: PullRequests) -> OrderPro def from_source(cls, source: ProgressSource) -> OrderProgress: members = sorted(cls.members_with(FoundProgress), key=lambda member: member.precedence) try: - return next((progress for member in members for progress in member.observations(source)), - NoCommitProgress()) + return next((progress for member in members + for progress in member.observations(source)), NoCommitProgress()) except subprocess.TimeoutExpired as unread: return UnreadProgress(str(unread)) @@ -426,7 +426,7 @@ def distance_readings(self) -> DistanceReadings: """The order's distance readings; progress that read no commit has none.""" return DistanceReadings() - def advance(self, todo: Todo, now: float) -> Todo: + def advance(self, todo: Todo, ordered: float, now: float) -> Todo: """Progress with no stall consequence leaves its order unchanged.""" return todo @@ -470,6 +470,10 @@ class Stalling: seconds: ClassVar[int] = 600 + @property + @abstractmethod + def state(self) -> str: ... + @abstractmethod def last_move(self, ordered: float) -> float: ... @@ -483,8 +487,8 @@ def stall_at(self, ordered: float) -> float: def stalled(self, ordered: float, now: float) -> bool: return now >= self.stall_at(ordered) - def advance(self, todo: Todo, now: float) -> Todo: - if not self.stalled(todo.order.at.recorded_at("An order"), now): + def advance(self, todo: Todo, ordered: float, now: float) -> Todo: + if not self.stalled(ordered, now): return todo owner, actor = todo.held(), todo.creator_identity reason = StrikeReason(f"{owner.owner} stalled on {todo.id}: {self.stall_reason()}") @@ -2327,7 +2331,7 @@ def advance_stalled(self, storage: StorageRule) -> tuple[Todo, ...]: with self.writing() as db: current = self._current(db, todo.id) current.require_result(todo) - result = progress.advance(current, observed_at) + result = progress.advance(current, source.since, observed_at) if result != current: result.upsert(db) changed.append(result) From a2cce8fecb23c765489d9bc5036933a5a5eafb42 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Mon, 19 Oct 2026 15:27:29 -0400 Subject: [PATCH 05/18] P717 git filters push reflog entries; no subject string compare Co-Authored-By: Claude Opus 5.5 --- src/agent_comms/todos.py | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/src/agent_comms/todos.py b/src/agent_comms/todos.py index 705cb6a8a7..92e9043b73 100644 --- a/src/agent_comms/todos.py +++ b/src/agent_comms/todos.py @@ -580,11 +580,11 @@ def pushes(cls, source: ProgressSource) -> Iterator[Self]: """A push-ref update has its own time, not the commit's author time.""" branch = source.git("branch", "--show-current").strip() for row in source.git("reflog", "show", "--all", "--date=unix", - f"--since={source.git_since}", - "--format=%gD%x09%H%x09%gs").splitlines(): - selector, sha, subject = row.split("\t", 2) + f"--since={source.git_since}", "--grep-reflog=update by push", + "--format=%gD%x09%H").splitlines(): + selector, sha = row.split("\t") reference, stamp = selector.rsplit("@{", 1) - if reference == f"refs/remotes/origin/{branch}" and subject == "update by push": + if reference == f"refs/remotes/origin/{branch}": yield cls(sha[:9], at=float(stamp.removesuffix("}"))) @classmethod From 69be1e0e3cf3be11d827fa107e307512058d104d Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Mon, 19 Oct 2026 16:01:43 -0400 Subject: [PATCH 06/18] P717 Core's stall watcher strikes and releases on its own declared identity StallWatcherIncarnation joins the thread-identity family as a birthless member (wire birth -2.0); deleted and unrecorded members now share BirthlessIncarnation, which owns the mark-to-member decode instead of an if-chain in the record. StallStrikeTodo/StallReleaseTodo take their permission from StallWatcherPermission; commander strikes keep CommanderPermission. advance() records the watcher, not the creator. Co-Authored-By: Claude Opus 5.5 --- src/agent_comms/thread_identity.py | 66 +++++++++++++++++++----------- src/agent_comms/todos.py | 42 +++++++++++++++---- tests/test_stall_watcher_strike.py | 39 ++++++++++++++++++ 3 files changed, 115 insertions(+), 32 deletions(-) create mode 100644 tests/test_stall_watcher_strike.py diff --git a/src/agent_comms/thread_identity.py b/src/agent_comms/thread_identity.py index 366cda2f82..71095659c8 100644 --- a/src/agent_comms/thread_identity.py +++ b/src/agent_comms/thread_identity.py @@ -162,52 +162,73 @@ def lookup(self) -> str: return hashlib.sha256(identity).hexdigest()[:32] -@dataclass(frozen=True) -class DeletedThreadIncarnation(ThreadIncarnation): - """A deleted participant retains only its name, never a lookup capability.""" +class BirthlessIncarnation(ThreadIncarnation): + """A member whose wire birth is a declared mark, not recorded evidence; never a lookup. - WIRE_BIRTH: ClassVar[float] = -1.0 + Each member declares its mark in ``WIRE_BIRTH``; the record decodes a mark to its member.""" + + WIRE_BIRTH: ClassVar[float] + members: ClassVar[dict[float, type[BirthlessIncarnation]]] = {} + + def __init_subclass__(cls, **kwargs) -> None: + super().__init_subclass__(**kwargs) + if cls.WIRE_BIRTH in BirthlessIncarnation.members: + raise TypeError(f"Wire birth {cls.WIRE_BIRTH} is already declared") + BirthlessIncarnation.members[cls.WIRE_BIRTH] = cls @property def wire_birth(self) -> float: return self.WIRE_BIRTH + def require_recorded(self) -> RecordedThreadIncarnation: + raise ValueError("Thread incarnation requires a positive recorded birth") + + def renamed(self, name: str) -> Self: + return type(self)(name) + + +@dataclass(frozen=True) +class DeletedThreadIncarnation(BirthlessIncarnation): + """A deleted participant retains only its name, never a lookup capability.""" + + WIRE_BIRTH: ClassVar[float] = -1.0 + def current(self, snapshot: RegistryProvenance) -> bool: return snapshot.canonical_name(self.name) not in snapshot.threads def resolved(self, snapshot: RegistryProvenance) -> DeletedThreadIncarnation: return self - def require_recorded(self) -> RecordedThreadIncarnation: - raise ValueError("Thread incarnation requires a positive recorded birth") - @property def birth_description(self) -> str: return "deleted declaration" - def renamed(self, name: str) -> DeletedThreadIncarnation: - return DeletedThreadIncarnation(name) - @dataclass(frozen=True) -class UnrecordedThreadIncarnation(ThreadIncarnation): +class UnrecordedThreadIncarnation(BirthlessIncarnation): """Historical zero-birth evidence cannot select a live recipient.""" WIRE_BIRTH: ClassVar[float] = 0.0 @property - def wire_birth(self) -> float: - return self.WIRE_BIRTH + def birth_description(self) -> str: + return "birth unrecorded" - def require_recorded(self) -> RecordedThreadIncarnation: - raise ValueError("Thread incarnation requires a positive recorded birth") + +@dataclass(frozen=True) +class StallWatcherIncarnation(BirthlessIncarnation): + """Core's own stall watcher: it strikes and releases stalled orders, and is never a thread.""" + + WIRE_BIRTH: ClassVar[float] = -2.0 + NAME: ClassVar[str] = "core-stall-watcher" + + @classmethod + def watcher(cls) -> StallWatcherIncarnation: + return cls(cls.NAME) @property def birth_description(self) -> str: - return "birth unrecorded" - - def renamed(self, name: str) -> UnrecordedThreadIncarnation: - return UnrecordedThreadIncarnation(name) + return "Core's stall watcher" @dataclass(frozen=True, kw_only=True) @@ -273,11 +294,8 @@ class ThreadIncarnationRecord: created_at: float def incarnation(self) -> ThreadIncarnation: - if self.created_at == DeletedThreadIncarnation.WIRE_BIRTH: - return DeletedThreadIncarnation(self.name) - if self.created_at == UnrecordedThreadIncarnation.WIRE_BIRTH: - return UnrecordedThreadIncarnation(self.name) - return RecordedThreadIncarnation(self.name, self.created_at) + member = BirthlessIncarnation.members.get(self.created_at) + return RecordedThreadIncarnation(self.name, self.created_at) if member is None else member(self.name) @classmethod def decode(cls, name: str, birth: float) -> ThreadIncarnation: diff --git a/src/agent_comms/todos.py b/src/agent_comms/todos.py index 92e9043b73..9fafe78d6b 100644 --- a/src/agent_comms/todos.py +++ b/src/agent_comms/todos.py @@ -35,7 +35,7 @@ from .channel_targets import CommandTarget from .goals import CountCheckResult, OrderCheck, ProblemGoalOrder from .recorded_time import EventTime, RecordedEventTime, UnrecordedEventTime -from .thread_identity import RecordedThreadIncarnation, ThreadIncarnation +from .thread_identity import RecordedThreadIncarnation, StallWatcherIncarnation, ThreadIncarnation from .threads import Thread from .typed_table import Column, Index, JsonStorage, TypedStore, TypedTable from .standing_rules import UndatedStandingRule @@ -490,11 +490,11 @@ def stalled(self, ordered: float, now: float) -> bool: def advance(self, todo: Todo, ordered: float, now: float) -> Todo: if not self.stalled(ordered, now): return todo - owner, actor = todo.held(), todo.creator_identity + owner, actor = todo.held(), StallWatcherIncarnation.watcher() reason = StrikeReason(f"{owner.owner} stalled on {todo.id}: {self.stall_reason()}") at = RecordedEventTime(now) - struck = StrikeTodo(todo.revision, actor, owner, reason.text, at).apply(todo) - return ReleaseTodo(struck.revision, actor, owner, reason, at).apply(struck) + struck = StallStrikeTodo(todo.revision, actor, owner, reason.text, at).apply(todo) + return StallReleaseTodo(struck.revision, actor, owner, reason, at).apply(struck) def stalled_between(self, ordered: float, since: float, now: float) -> bool: return since < self.stall_at(ordered) <= now @@ -1696,8 +1696,24 @@ def apply(self, current: Todo) -> Todo: revision=current.revision + 1) +class CommanderPermission: + """Strikes and releases are the order's commander's to make.""" + actor: ThreadIncarnation + + def require_permitted(self, current: Todo) -> None: + current.require_commander(self.actor) + + +class StallWatcherPermission(CommanderPermission): + """Core's stall watcher makes these on its own declared identity, never the commander's.""" + + def require_permitted(self, current: Todo) -> None: + if self.actor != StallWatcherIncarnation.watcher(): + raise TodoConflict(current) + + @dataclass(frozen=True) -class StrikeTodo(TodoOperation): +class StrikeTodo(CommanderPermission, TodoOperation): actor: ThreadIncarnation previous: Assignment reason: str @@ -1705,7 +1721,7 @@ class StrikeTodo(TodoOperation): def apply(self, current: Todo) -> Todo: current.require_revision(self.expected_revision) - current.require_commander(self.actor) + self.require_permitted(current) current.require_assignment(self.previous) _text(self.reason, "Strike reason", 2000) StrikeReason(self.reason) @@ -1869,7 +1885,7 @@ def apply(self, current: Todo) -> Todo: @dataclass(frozen=True) -class ReleaseTodo(TodoOperation): +class ReleaseTodo(CommanderPermission, TodoOperation): """The commander takes an open problem off its unit; the order and its strikes stay.""" actor: ThreadIncarnation previous: Assignment @@ -1878,12 +1894,22 @@ class ReleaseTodo(TodoOperation): def apply(self, current: Todo) -> Todo: current.require_revision(self.expected_revision) - current.require_commander(self.actor) + self.require_permitted(current) current.require_assignment(self.previous) return replace(current, assignment=None, released=(*current.released, self), revision=current.revision + 1) +@dataclass(frozen=True) +class StallStrikeTodo(StallWatcherPermission, StrikeTodo): + """A strike Core's stall watcher records when an order shows no movement.""" + + +@dataclass(frozen=True) +class StallReleaseTodo(StallWatcherPermission, ReleaseTodo): + """Core's stall watcher takes the stalled order off its unit.""" + + @dataclass(frozen=True) class CorrectCheckTodo(TodoOperation): """The commander replaces a wrong check of an open order. diff --git a/tests/test_stall_watcher_strike.py b/tests/test_stall_watcher_strike.py new file mode 100644 index 0000000000..72060790de --- /dev/null +++ b/tests/test_stall_watcher_strike.py @@ -0,0 +1,39 @@ +import importlib + +from agent_comms.change_report import FileLine +from agent_comms.goal_actions import CreateProblemGoalAction +from agent_comms.goals import CountCheckResult, OrderCheck +from agent_comms.recorded_time import RecordedEventTime +from agent_comms.thread_identity import StallWatcherIncarnation +from agent_comms.todos import MechanicalOrderRun, PendingProblemOrder, StallReleaseTodo, StallStrikeTodo +from agent_comms.tools import CommsTodoCreateTool + + +def test_stalled_order_is_struck_by_core_stall_watcher(commander, commander_script, monkeypatch): + comms, caller = commander + from agent_comms.commander import standing + standing = importlib.reload(standing) + comms.registry.declare(comms.registry.new_thread("spare", frozenset({'comms'}), str(comms.root), created_at=3.0)) + order = PendingProblemOrder(id="order-P1", owner=FileLine(__file__, 1), run=MechanicalOrderRun(), + check=OrderCheck(str(comms.root), "printf 1", CountCheckResult(1)), edits=("Repair the owner",), + banned=("No second store",), report="Report the result", + at=RecordedEventTime(100), first_call_deadline=RecordedEventTime(220)) + CommsTodoCreateTool(owner="worker", order=CreateProblemGoalAction(problem="P1", order=order, + request="request", deadline=RecordedEventTime(1000), commander="lead", + repo="OpenHCSDev/agent-comms", text="Fix P1")).apply(comms, caller) + changed = comms.goals.todos.advance_stalled(comms.registry.storage_rule) + todo = comms.goals.todos.get("P1") + assert [t.id for t in changed] == ["P1"] + # read back from the store: the strike keeps its kind and the watcher as its actor + assert [type(s) for s in todo.strikes] == [StallStrikeTodo] + assert [type(r) for r in todo.released] == [StallReleaseTodo] + assert todo.strikes[0].actor == todo.released[0].actor == StallWatcherIncarnation.watcher() + assert todo.holder is None and todo.live_strikes("worker") + # the one stubbed read: turn-digest activity marks worker and spare as recently active + monkeypatch.setattr(standing.WheelSpin, "idle", classmethod( + lambda cls, c, cmd: [c.registry.require("worker"), c.registry.require("spare")])) + found = standing.UnownedWheelSpin.violations(comms, "lead") + assert [(f.todo.id, f.unit) for f in found] == [("P1", "spare")] + for rule in found: + rule.enforce(comms, "lead") + assert comms.goals.todos.get("P1").holder == "spare" From 409d46c84d640f3d45bae7ca63f0a6c200df1253 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Mon, 19 Oct 2026 16:28:46 -0400 Subject: [PATCH 07/18] P717 the stall watcher hands a released order to an idle unit in the same pass After Core's stall watcher struck and released a stalled order, nothing gave the order to anyone until a commander read its standing rules. The watcher pass now runs UnownedWheelSpin's own pairing on the orders it just released, for each order's commander, so the reassign happens in the same pass with no commander turn. WheelSpin and UnownedWheelSpin move to commander/wheel_spin.py because commander/standing.py imports order.py, which opens the commander's configured bus on import; the connector must not do that. TodoStore loses its watch loop; the rule owns it. StallWatcherIncarnation's name now derives from its class name through the declared-name conversion DeclaredFamily already used, shared as declared_words(); the hand-written "core-stall-watcher" literal is gone. Co-Authored-By: Claude Opus 5.5 --- src/agent_comms/commander/standing.py | 67 +--------------- src/agent_comms/commander/wheel_spin.py | 100 ++++++++++++++++++++++++ src/agent_comms/connector.py | 4 +- src/agent_comms/declared_family.py | 11 ++- src/agent_comms/thread_identity.py | 5 +- src/agent_comms/todos.py | 8 -- tests/test_stall_watcher_strike.py | 24 +++--- 7 files changed, 126 insertions(+), 93 deletions(-) create mode 100644 src/agent_comms/commander/wheel_spin.py diff --git a/src/agent_comms/commander/standing.py b/src/agent_comms/commander/standing.py index c17c1a032a..3d53452923 100644 --- a/src/agent_comms/commander/standing.py +++ b/src/agent_comms/commander/standing.py @@ -3,22 +3,19 @@ from __future__ import annotations import subprocess -import time from collections.abc import Iterator from dataclasses import dataclass from pathlib import Path from typing import ClassVar from ..comms import Comms -from ..goal_actions import OrderGoalAction -from ..standing_rules import (DatedStandingRule, ReasonedStandingRule, StandingRule, - UndatedStandingRule) +from ..standing_rules import (DatedStandingRule, ReasonedStandingRule, UndatedStandingRule) from ..goals import Goal -from ..threads import Thread -from ..todos import FlatTrend, OrderProgress, PullRequests, StrikeTodo, Todo +from ..todos import FlatTrend, OrderProgress, PullRequests, StrikeTodo from ..turn_digests import TurnDigestRead, TurnsSince from . import capacity, judge from .config import CommanderConfig, Repo +from .wheel_spin import WheelSpin CONFIG = CommanderConfig.installed() @@ -60,37 +57,6 @@ def line(self) -> str: " -> order its next step or mark it done") -@dataclass(frozen=True) -class WheelSpin(StandingRule, declared_name=None): - """An order is not getting closer to done, or a unit sits with no order; Core gives the - order to a unit that will move it, with no deadline and no commander turn involved.""" - - todo: Todo - - recent: ClassVar[int] = 10 * 60 - - @classmethod - def idle(cls, comms: Comms, commander: str) -> list[Thread]: - """Units with a native turn in the last ten minutes and no order, parked threads aside.""" - held = {todo.holder for todo in cls.todos(comms, commander)} - excluded = held | CONFIG.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).native()), - key=lambda thread: thread.name) - - def assign(self, comms: Comms, commander: str, unit: Thread) -> None: - """The order sent again to ``unit`` through its Goal, with the order's own text.""" - todo = comms.goals.todos.get(self.todo.id) - 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)) - comms.messaging.send(commander, unit.name, f"{self.line()}\n{order.instructions(todo)}") - - @dataclass(frozen=True) class FlatWheelSpin(WheelSpin, DatedStandingRule): """Two distance readings with no fall while the unit kept working.""" @@ -129,33 +95,6 @@ def enforce(self, comms: Comms, commander: str) -> None: self.assign(comms, commander, unit) -@dataclass(frozen=True) -class UnownedWheelSpin(WheelSpin, UndatedStandingRule): - """A unit holds no next order while the board has unowned rows.""" - - unit: str - - @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, found = cls.idle(comms, commander), [] - for todo in rows: # a unit struck on a problem never gets that problem back - 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)) - - @dataclass(frozen=True) class ForkNow(UndatedStandingRule): """The load on the asked Comms needs more live forks than there are.""" diff --git a/src/agent_comms/commander/wheel_spin.py b/src/agent_comms/commander/wheel_spin.py new file mode 100644 index 0000000000..90340eba3f --- /dev/null +++ b/src/agent_comms/commander/wheel_spin.py @@ -0,0 +1,100 @@ +"""Core gives a stalled or unowned order to an idle unit; the stall watcher and the commander's +standing rules share it, so this module loads nothing that opens a bus on import.""" + +from __future__ import annotations + +import asyncio +import time +from collections.abc import Iterable +from dataclasses import dataclass +from typing import ClassVar + +from ..comms import Comms +from ..coordinator import Coordination +from ..goal_actions import OrderGoalAction +from ..standing_rules import StandingRule, UndatedStandingRule +from ..threads import Thread +from ..todos import Todo +from ..turn_digests import TurnDigestRead, TurnsSince +from .config import CommanderConfig + + +@dataclass(frozen=True) +class WheelSpin(StandingRule, declared_name=None): + """An order is not getting closer to done, or a unit sits with no order; Core gives the + order to a unit that will move it, with no deadline and no commander turn involved.""" + + todo: Todo + + recent: ClassVar[int] = 10 * 60 + + @classmethod + def idle(cls, comms: Comms, commander: str) -> list[Thread]: + """Units with a native turn in the last ten minutes and no order, parked threads aside.""" + held = {todo.holder for todo in cls.todos(comms, 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).native()), + key=lambda thread: thread.name) + + def assign(self, comms: Comms, commander: str, unit: Thread) -> None: + """The order sent again to ``unit`` through its Goal, with the order's own text.""" + todo = comms.goals.todos.get(self.todo.id) + 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)) + comms.messaging.send(commander, unit.name, f"{self.line()}\n{order.instructions(todo)}") + + + +@dataclass(frozen=True) +class UnownedWheelSpin(WheelSpin, UndatedStandingRule): + """A unit holds no next order while the board has unowned rows.""" + + unit: str + + @classmethod + def violations(cls, comms: Comms, commander: str) -> tuple[UnownedWheelSpin, ...]: + return cls.paired(comms, commander, comms.goals.todos.for_commander( + comms.registry.require(commander).incarnation)) + + @classmethod + def paired(cls, comms: Comms, commander: str, todos: Iterable[Todo]) -> tuple[UnownedWheelSpin, ...]: + """Each unowned pending row with the first idle unit not struck on it.""" + rows = (todo for todo in todos if todo.pending and not todo.holder) + idle, found = cls.idle(comms, commander), [] + for todo in rows: # a unit struck on a problem never gets that problem back + 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)) + + @classmethod + def after_stall(cls, comms: Comms) -> None: + """One stall watcher pass: strike and release stalled orders, then give each released + order to an idle unit for its own commander, with no commander turn involved.""" + released = comms.goals.todos.advance_stalled(comms.registry.storage_rule) + for commander in {todo.creator_identity.name for todo in released}: + mine = (todo for todo in released if todo.creator_identity.name == commander) + for rule in cls.paired(comms, commander, mine): + rule.enforce(comms, commander) + + @classmethod + async def watch(cls, comms: Comms) -> None: + """The running Core service makes one stall pass a minute.""" + while True: + await Coordination.run_worker(lambda: cls.after_stall(comms)) + await asyncio.sleep(60) diff --git a/src/agent_comms/connector.py b/src/agent_comms/connector.py index a6c0c43723..845cbc3ba2 100644 --- a/src/agent_comms/connector.py +++ b/src/agent_comms/connector.py @@ -49,6 +49,7 @@ RegisteredClient, digest, ) from .messages import Message +from .commander.wheel_spin import UnownedWheelSpin from .standing_rules import RulesDue from .todos import DirectiveText from .tool_caller import ThreadCaller @@ -383,8 +384,7 @@ def application(self) -> web.Application: async def watch_orders(self, app: web.Application) -> AsyncIterator[None]: """The running Core service advances stalled orders without a commander call.""" with self.opened_wire() as comms: - task = asyncio.create_task( - comms.goals.todos.watch_stalled(comms.registry.storage_rule)) + task = asyncio.create_task(UnownedWheelSpin.watch(comms)) try: yield finally: diff --git a/src/agent_comms/declared_family.py b/src/agent_comms/declared_family.py index f8e864842b..93e098dea5 100644 --- a/src/agent_comms/declared_family.py +++ b/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) diff --git a/src/agent_comms/thread_identity.py b/src/agent_comms/thread_identity.py index 71095659c8..d9622c802c 100644 --- a/src/agent_comms/thread_identity.py +++ b/src/agent_comms/thread_identity.py @@ -16,6 +16,7 @@ from .agent_backend import AgentBackend, RuntimeLocation from .channel_targets import Tag +from .declared_family import declared_words from .cli_options import JsonFlagRecord from .declared_family import DeclaredFamily from .errors import RelationViolationError @@ -220,11 +221,11 @@ class StallWatcherIncarnation(BirthlessIncarnation): """Core's own stall watcher: it strikes and releases stalled orders, and is never a thread.""" WIRE_BIRTH: ClassVar[float] = -2.0 - NAME: ClassVar[str] = "core-stall-watcher" @classmethod def watcher(cls) -> StallWatcherIncarnation: - return cls(cls.NAME) + """Named by its class: ``StallWatcherIncarnation`` is ``stall_watcher``.""" + return cls(declared_words(cls.__name__.removesuffix("Incarnation"))) @property def birth_description(self) -> str: diff --git a/src/agent_comms/todos.py b/src/agent_comms/todos.py index 9fafe78d6b..1e79aa4add 100644 --- a/src/agent_comms/todos.py +++ b/src/agent_comms/todos.py @@ -24,7 +24,6 @@ from .change_report import FileLine from .child_process import BoundedRun, ChildResult, ExitedOutcome -from .coordinator import Coordination from .command import Command from .declared_family import DeclaredFamily from .errors import ToolRefusedError @@ -2366,13 +2365,6 @@ def advance_stalled(self, storage: StorageRule) -> tuple[Todo, ...]: continue return tuple(changed) - async def watch_stalled(self, storage: StorageRule) -> None: - """The backend observes Git/PR changes; Stalling alone owns the deadline.""" - - while True: - await Coordination.run_worker(lambda: self.advance_stalled(storage)) - await asyncio.sleep(60) - def board_disagreements( self, commander: ThreadIncarnation, storage: StorageRule, ) -> Iterator[tuple[Todo, MergedProgress]]: diff --git a/tests/test_stall_watcher_strike.py b/tests/test_stall_watcher_strike.py index 72060790de..e41106c3dd 100644 --- a/tests/test_stall_watcher_strike.py +++ b/tests/test_stall_watcher_strike.py @@ -9,10 +9,10 @@ from agent_comms.tools import CommsTodoCreateTool -def test_stalled_order_is_struck_by_core_stall_watcher(commander, commander_script, monkeypatch): +def test_stall_pass_strikes_releases_and_reassigns_on_the_watcher_identity(commander, commander_script, monkeypatch): comms, caller = commander - from agent_comms.commander import standing - standing = importlib.reload(standing) + from agent_comms.commander import wheel_spin + wheel_spin = importlib.reload(wheel_spin) comms.registry.declare(comms.registry.new_thread("spare", frozenset({'comms'}), str(comms.root), created_at=3.0)) order = PendingProblemOrder(id="order-P1", owner=FileLine(__file__, 1), run=MechanicalOrderRun(), check=OrderCheck(str(comms.root), "printf 1", CountCheckResult(1)), edits=("Repair the owner",), @@ -21,19 +21,15 @@ def test_stalled_order_is_struck_by_core_stall_watcher(commander, commander_scri CommsTodoCreateTool(owner="worker", order=CreateProblemGoalAction(problem="P1", order=order, request="request", deadline=RecordedEventTime(1000), commander="lead", repo="OpenHCSDev/agent-comms", text="Fix P1")).apply(comms, caller) - changed = comms.goals.todos.advance_stalled(comms.registry.storage_rule) + # the one stubbed read: turn-digest activity marks worker and spare as recently active + monkeypatch.setattr(wheel_spin.WheelSpin, "idle", classmethod( + lambda cls, c, cmd: [c.registry.require("worker"), c.registry.require("spare")])) + wheel_spin.UnownedWheelSpin.after_stall(comms) todo = comms.goals.todos.get("P1") - assert [t.id for t in changed] == ["P1"] # read back from the store: the strike keeps its kind and the watcher as its actor assert [type(s) for s in todo.strikes] == [StallStrikeTodo] assert [type(r) for r in todo.released] == [StallReleaseTodo] assert todo.strikes[0].actor == todo.released[0].actor == StallWatcherIncarnation.watcher() - assert todo.holder is None and todo.live_strikes("worker") - # the one stubbed read: turn-digest activity marks worker and spare as recently active - monkeypatch.setattr(standing.WheelSpin, "idle", classmethod( - lambda cls, c, cmd: [c.registry.require("worker"), c.registry.require("spare")])) - found = standing.UnownedWheelSpin.violations(comms, "lead") - assert [(f.todo.id, f.unit) for f in found] == [("P1", "spare")] - for rule in found: - rule.enforce(comms, "lead") - assert comms.goals.todos.get("P1").holder == "spare" + assert StallWatcherIncarnation.watcher().name == "stall_watcher" + # the same pass gives the order to the idle unit not struck on it, never back to worker + assert todo.holder == "spare" and todo.live_strikes("worker") From 20f38c588b9603b98cfde9c1c616a85d9f90b05c Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Mon, 19 Oct 2026 16:34:05 -0400 Subject: [PATCH 08/18] P717 fix what the ratchet found in this branch's files Recorded births pick their member by membership, not by a None lookup. reissued() lives on the ProblemOrder base, so the wheel-spin assign needs no narrowing, and it traces the reissued order's own source line. Co-Authored-By: Claude Opus 5.5 --- src/agent_comms/commander/wheel_spin.py | 47 +++++++++++++++++-------- src/agent_comms/thread_identity.py | 10 +++--- src/agent_comms/todos.py | 8 ++--- 3 files changed, 42 insertions(+), 23 deletions(-) diff --git a/src/agent_comms/commander/wheel_spin.py b/src/agent_comms/commander/wheel_spin.py index 90340eba3f..bddb0c0d78 100644 --- a/src/agent_comms/commander/wheel_spin.py +++ b/src/agent_comms/commander/wheel_spin.py @@ -34,23 +34,35 @@ def idle(cls, comms: Comms, commander: str) -> list[Thread]: held = {todo.holder for todo in cls.todos(comms, 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).native()), - key=lambda thread: thread.name) + return sorted( + ( + thread + for thread in comms.registry.all_threads().values() + if thread.name not in excluded + and TurnDigestRead(comms.root, thread, bound).native() + ), + key=lambda thread: thread.name, + ) def assign(self, comms: Comms, commander: str, unit: Thread) -> None: """The order sent again to ``unit`` through its Goal, with the order's own text.""" todo = comms.goals.todos.get(self.todo.id) 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)) + 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=order.owner, + ), + ) comms.messaging.send(commander, unit.name, f"{self.line()}\n{order.instructions(todo)}") - @dataclass(frozen=True) class UnownedWheelSpin(WheelSpin, UndatedStandingRule): """A unit holds no next order while the board has unowned rows.""" @@ -59,11 +71,16 @@ class UnownedWheelSpin(WheelSpin, UndatedStandingRule): @classmethod def violations(cls, comms: Comms, commander: str) -> tuple[UnownedWheelSpin, ...]: - return cls.paired(comms, commander, comms.goals.todos.for_commander( - comms.registry.require(commander).incarnation)) + return cls.paired( + comms, + commander, + comms.goals.todos.for_commander(comms.registry.require(commander).incarnation), + ) @classmethod - def paired(cls, comms: Comms, commander: str, todos: Iterable[Todo]) -> tuple[UnownedWheelSpin, ...]: + def paired( + cls, comms: Comms, commander: str, todos: Iterable[Todo] + ) -> tuple[UnownedWheelSpin, ...]: """Each unowned pending row with the first idle unit not struck on it.""" rows = (todo for todo in todos if todo.pending and not todo.holder) idle, found = cls.idle(comms, commander), [] @@ -76,8 +93,10 @@ def paired(cls, comms: Comms, commander: str, todos: Iterable[Todo]) -> tuple[Un 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}") + 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)) diff --git a/src/agent_comms/thread_identity.py b/src/agent_comms/thread_identity.py index d9622c802c..0598c7af10 100644 --- a/src/agent_comms/thread_identity.py +++ b/src/agent_comms/thread_identity.py @@ -16,9 +16,8 @@ from .agent_backend import AgentBackend, RuntimeLocation from .channel_targets import Tag -from .declared_family import declared_words from .cli_options import JsonFlagRecord -from .declared_family import DeclaredFamily +from .declared_family import DeclaredFamily, declared_words from .errors import RelationViolationError from .field_codec import FieldCodec, WireValue from .response_policy import CollectivePolicy, InformationalPolicy, ResponsePolicy @@ -171,7 +170,7 @@ class BirthlessIncarnation(ThreadIncarnation): WIRE_BIRTH: ClassVar[float] members: ClassVar[dict[float, type[BirthlessIncarnation]]] = {} - def __init_subclass__(cls, **kwargs) -> None: + def __init_subclass__(cls, **kwargs: object) -> None: super().__init_subclass__(**kwargs) if cls.WIRE_BIRTH in BirthlessIncarnation.members: raise TypeError(f"Wire birth {cls.WIRE_BIRTH} is already declared") @@ -295,8 +294,9 @@ class ThreadIncarnationRecord: created_at: float def incarnation(self) -> ThreadIncarnation: - member = BirthlessIncarnation.members.get(self.created_at) - return RecordedThreadIncarnation(self.name, self.created_at) if member is None else member(self.name) + if self.created_at in BirthlessIncarnation.members: + return BirthlessIncarnation.members[self.created_at](self.name) + return RecordedThreadIncarnation(self.name, self.created_at) @classmethod def decode(cls, name: str, birth: float) -> ThreadIncarnation: diff --git a/src/agent_comms/todos.py b/src/agent_comms/todos.py index 1e79aa4add..cc2ad2e739 100644 --- a/src/agent_comms/todos.py +++ b/src/agent_comms/todos.py @@ -1148,6 +1148,10 @@ class ProblemOrder(DeclaredFamily, affix="ProblemOrder"): finding: ProblemFinding = field(default_factory=UnlocatedProblemFinding, metadata={"wire_omit_default": True}) + def reissued(self, at: float) -> PendingProblemOrder: + """This order sent again at ``at`` to a new owner, its first-call window kept.""" + raise ValueError("Only an undelivered order can change owner.") + @classmethod def pull_request_ids(cls, body: str) -> tuple[str, ...]: return tuple(line.removeprefix(cls.pull_request_label).strip() @@ -1248,10 +1252,6 @@ def __post_init__(self): def require_transfer(self, todo: Todo) -> None: raise ValueError("Only an undelivered order can change owner.") - def reissued(self, at: float) -> PendingProblemOrder: - """This order sent again at ``at`` to a new owner, its first-call window kept.""" - raise ValueError("Only an undelivered order can change owner.") - def untouched(self, touched: Sequence[str]) -> int: """The numbered edits naming a file:line where ``touched`` changed none of those files.""" def changed(name: str) -> bool: From 3a1841349a9543abdda1b7e1897735b3b9f8a8f9 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Mon, 19 Oct 2026 17:02:49 -0400 Subject: [PATCH 09/18] P717 stall watcher: Core owns pairing and parked set, records failed passes, assigns as the watcher Co-Authored-By: Claude Opus 5.5 --- src/agent_comms/cli_commands.py | 20 ++++ src/agent_comms/commander/config.py | 1 - src/agent_comms/commander/standing.py | 2 +- src/agent_comms/connector.py | 4 +- src/agent_comms/goal_actions.py | 20 +++- .../wheel_spin.py => stall_watcher.py} | 61 ++++++++--- src/agent_comms/todos.py | 102 +++++++++++++----- tests/test_stall_watcher_strike.py | 45 +++++++- 8 files changed, 201 insertions(+), 54 deletions(-) rename src/agent_comms/{commander/wheel_spin.py => stall_watcher.py} (66%) diff --git a/src/agent_comms/cli_commands.py b/src/agent_comms/cli_commands.py index c69ae0b9d8..8df0c0c925 100644 --- a/src/agent_comms/cli_commands.py +++ b/src/agent_comms/cli_commands.py @@ -645,6 +645,11 @@ class RegisteredResult(CliResult): registered: str +@dataclass(frozen=True) +class ParkedResult(CliResult): + parked: tuple[str, ...] + + @dataclass(frozen=True) class HeartbeatResult(CliResult): heartbeat: str @@ -1478,6 +1483,21 @@ def apply(self, ctx: Comms) -> CliResult: for harness in Harness.present(home)}) +@dataclass(frozen=True, kw_only=True) +class ParkCliCommand(CliCommand): + help = "Keep Core's stall watcher from giving orders to a thread, or allow it again with --off" + name: str = option("--name") + off: bool = option("--off", default=False) + + def apply(self, ctx: Comms) -> CliResult: + todos = ctx.goals.todos + if self.off: + todos.unpark(self.name) + else: + todos.park(ctx.registry.require(self.name).name) + return ParkedResult(tuple(sorted(todos.parked()))) + + @dataclass(frozen=True, kw_only=True) class HeartbeatCliCommand(CliCommand): help = "Mark thread running" diff --git a/src/agent_comms/commander/config.py b/src/agent_comms/commander/config.py index f6607f7d09..d2d813cf21 100644 --- a/src/agent_comms/commander/config.py +++ b/src/agent_comms/commander/config.py @@ -62,7 +62,6 @@ class CommanderConfig(CommanderRole, BaseModel): worktrees: Path scratch: Path disk: Path - parked: frozenset[str] = frozenset() @staticmethod def path(home: Path | None = None) -> Path: diff --git a/src/agent_comms/commander/standing.py b/src/agent_comms/commander/standing.py index 3d53452923..8a2dce4532 100644 --- a/src/agent_comms/commander/standing.py +++ b/src/agent_comms/commander/standing.py @@ -15,7 +15,7 @@ from ..turn_digests import TurnDigestRead, TurnsSince from . import capacity, judge from .config import CommanderConfig, Repo -from .wheel_spin import WheelSpin +from ..stall_watcher import WheelSpin CONFIG = CommanderConfig.installed() diff --git a/src/agent_comms/connector.py b/src/agent_comms/connector.py index a06b097556..300833a204 100644 --- a/src/agent_comms/connector.py +++ b/src/agent_comms/connector.py @@ -49,7 +49,7 @@ RegisteredClient, digest, ) from .messages import Message -from .commander.wheel_spin import UnownedWheelSpin +from .stall_watcher import StallWheelSpin from .todos import DirectiveText from .tool_caller import ThreadCaller from .tools import CommsChannelsTool, ReadTool, ToolRequest @@ -377,7 +377,7 @@ def application(self) -> web.Application: async def watch_orders(self, app: web.Application) -> AsyncIterator[None]: """The running Core service advances stalled orders without a commander call.""" with self.opened_wire() as comms: - task = asyncio.create_task(UnownedWheelSpin.watch(comms)) + task = asyncio.create_task(StallWheelSpin.watch(comms)) try: yield finally: diff --git a/src/agent_comms/goal_actions.py b/src/agent_comms/goal_actions.py index e5ca5b006f..c33c6da20f 100644 --- a/src/agent_comms/goal_actions.py +++ b/src/agent_comms/goal_actions.py @@ -39,7 +39,9 @@ from .progress_claims import ProgressClaim from .recorded_time import EventTime from .threads import Thread -from .todos import GoalRef, PendingProblemOrder, SendBackVerdict, Todo, TodoOperation +from .thread_identity import StallWatcherIncarnation, ThreadIncarnation +from .todos import (AssignTodo, GoalRef, PendingProblemOrder, SendBackVerdict, StallAssignTodo, Todo, + TodoOperation) if TYPE_CHECKING: from .goal_management import Goals @@ -542,16 +544,30 @@ 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}) + assignment: ClassVar[type[AssignTodo]] = AssignTodo + + def assigner(self, ctx: GoalActionContext) -> ThreadIncarnation: + return ctx.goals.registry.require(self.commander).incarnation 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(ctx), operation=self.assignment, generation=generation, goal=goal, deadline=self.deadline, order=self.order, traced=self.traced) +@dataclass(frozen=True, kw_only=True) +class StallOrderGoalAction(OrderGoalAction): + """Core's stall watcher gives a released order of ``commander`` to this thread.""" + assignment: ClassVar[type[AssignTodo]] = StallAssignTodo + + def assigner(self, ctx: GoalActionContext) -> ThreadIncarnation: + return StallWatcherIncarnation.watcher() + + @dataclass(frozen=True, kw_only=True) class CreateProblemGoalAction(ProblemOrderGoalAction): repo: str diff --git a/src/agent_comms/commander/wheel_spin.py b/src/agent_comms/stall_watcher.py similarity index 66% rename from src/agent_comms/commander/wheel_spin.py rename to src/agent_comms/stall_watcher.py index bddb0c0d78..d655348e23 100644 --- a/src/agent_comms/commander/wheel_spin.py +++ b/src/agent_comms/stall_watcher.py @@ -7,16 +7,15 @@ import time from collections.abc import Iterable from dataclasses import dataclass -from typing import ClassVar +from typing import ClassVar, Self -from ..comms import Comms -from ..coordinator import Coordination -from ..goal_actions import OrderGoalAction -from ..standing_rules import StandingRule, UndatedStandingRule -from ..threads import Thread -from ..todos import Todo -from ..turn_digests import TurnDigestRead, TurnsSince -from .config import CommanderConfig +from .comms import Comms +from .coordinator import Coordination +from .goal_actions import OrderGoalAction, StallOrderGoalAction +from .standing_rules import StandingRule, UndatedStandingRule +from .threads import Thread +from .todos import StallPassFailed, Todo, TodoConflict +from .turn_digests import TurnDigestRead, TurnsSince @dataclass(frozen=True) @@ -27,12 +26,13 @@ class WheelSpin(StandingRule, declared_name=None): todo: Todo recent: ClassVar[int] = 10 * 60 + order_action: ClassVar[type[OrderGoalAction]] = OrderGoalAction @classmethod def idle(cls, comms: Comms, commander: str) -> list[Thread]: """Units with a native turn in the last ten minutes and no order, parked threads aside.""" held = {todo.holder for todo in cls.todos(comms, commander)} - excluded = held | CommanderConfig.installed().parked | {commander, ""} + excluded = held | comms.goals.todos.parked() | {commander, ""} bound = TurnsSince(time.time() - cls.recent) return sorted( ( @@ -50,7 +50,7 @@ def assign(self, comms: Comms, commander: str, unit: Thread) -> None: order = todo.order.reissued(time.time()) comms.goals.update_goal( unit.name, - OrderGoalAction( + self.order_action( problem=todo.id, expected_revision=todo.revision, order=order, @@ -101,6 +101,13 @@ def line(self) -> str: def enforce(self, comms: Comms, commander: str) -> None: self.assign(comms, commander, comms.registry.require(self.unit)) + +@dataclass(frozen=True) +class StallWheelSpin(UnownedWheelSpin, declared_name=None): + """The stall watcher's own pairing: Core, not the commander, makes the assignment.""" + + order_action: ClassVar[type[OrderGoalAction]] = StallOrderGoalAction + @classmethod def after_stall(cls, comms: Comms) -> None: """One stall watcher pass: strike and release stalled orders, then give each released @@ -109,11 +116,37 @@ def after_stall(cls, comms: Comms) -> None: for commander in {todo.creator_identity.name for todo in released}: mine = (todo for todo in released if todo.creator_identity.name == commander) for rule in cls.paired(comms, commander, mine): - rule.enforce(comms, commander) + try: + rule.enforce(comms, commander) + except TodoConflict: + # A concurrent order change owns the row; the next pass sees its new state. + continue + comms.goals.todos.stall_pass_succeeded() @classmethod async def watch(cls, comms: Comms) -> None: - """The running Core service makes one stall pass a minute.""" + """The running Core service makes one stall pass a minute; a pass that raises is + recorded for the board and the next pass still runs.""" while True: - await Coordination.run_worker(lambda: cls.after_stall(comms)) + try: + await Coordination.run_worker(lambda: cls.after_stall(comms)) + except Exception as error: # noqa: BLE001 - recorded on the board, never swallowed + comms.goals.todos.stall_pass_failed( + StallPassFailed(time.time(), f"{type(error).__name__}: {error}")) await asyncio.sleep(60) + + +@dataclass(frozen=True) +class StallWatcherFailed(UndatedStandingRule): + """A stall watcher pass raised and no pass has succeeded since.""" + + failed: StallPassFailed + + @classmethod + def violations(cls, comms: Comms, commander: str) -> tuple[Self, ...]: + return tuple(map(cls, comms.goals.todos.stall_pass_failures())) + + def line(self) -> str: + at = time.strftime("%H:%M", time.localtime(self.failed.at)) + return (f"STALL WATCHER FAILED at {at}: {self.failed.error} -> fix the cause; the next" + " pass runs within a minute") diff --git a/src/agent_comms/todos.py b/src/agent_comms/todos.py index cc2ad2e739..6d1f4e16bc 100644 --- a/src/agent_comms/todos.py +++ b/src/agent_comms/todos.py @@ -1479,6 +1479,19 @@ class HuntRecord(TypedTable): hunt: Hunt +@dataclass(frozen=True) +class ParkedThread(TypedTable): + """A thread Core's stall watcher never gives an order to.""" + name: str = field(metadata={"sql": Column(primary_key=True)}) + + +@dataclass(frozen=True) +class StallPassFailed(TypedTable): + """A stall watcher pass that raised; the board shows it until a later pass succeeds.""" + at: float = field(metadata={"sql": Column(primary_key=True)}) + error: str + + class TodoError(ToolRefusedError): """A todo request is invalid or its persisted state cannot be trusted.""" @@ -1554,9 +1567,8 @@ def __post_init__(self) -> None: _text(self.parent, "Thread name", 128) try: self.owner_identity.require_recorded() - self.parent_identity.require_recorded() except ValueError as error: - raise TodoError("Assignment requires recorded owner and parent incarnations") from error + raise TodoError("Assignment requires a recorded owner incarnation") from error _text(self.generation, "Assignment generation", 128) @property @@ -1565,7 +1577,8 @@ def owner_identity(self) -> ThreadIncarnation: @property def parent_identity(self) -> ThreadIncarnation: - return RecordedThreadIncarnation(self.parent, self.parent_created) + """Who made the assignment: the order's commander, or Core's stall watcher.""" + return ThreadIncarnation.from_wire({"name": self.parent, "created_at": self.parent_created}) def owned_by(self, actor: ThreadIncarnation, generation: str | None) -> bool: return (self.owner_identity, self.generation) == (actor, generation) @@ -1673,8 +1686,25 @@ def record(self, goals: Goals, problem: str) -> Todo: def apply(self, current: Todo) -> Todo: ... +class CommanderPermission: + """Assignments, strikes and releases are the order's commander's to make.""" + actor: ThreadIncarnation + + def require_permitted(self, current: Todo) -> None: + current.require_commander(self.actor) + + +class StallWatcherPermission(CommanderPermission): + """Core's stall watcher makes these on its own declared identity, never the commander's.""" + + def require_permitted(self, current: Todo) -> None: + if self.actor != StallWatcherIncarnation.watcher(): + raise TodoConflict(current) + + @dataclass(frozen=True) -class AssignTodo(TodoOperation): +class AssignTodo(CommanderPermission, TodoOperation): + """The order's commander gives the problem to a unit.""" proposed: Assignment deadline: EventTime order: PendingProblemOrder @@ -1688,27 +1718,20 @@ def apply(self, current: Todo) -> Todo: proposed, self.deadline, self.order, self.goal): return current current.require_revision(self.expected_revision) - current.require_commander(proposed.parent_identity) + self.require_permitted(current) assigned = (current if current.assignment == proposed else current.state.assign(current, proposed, self.traced)) return replace(assigned, deadline=self.deadline, order=self.order, goal=self.goal, revision=current.revision + 1) + @property + def actor(self) -> ThreadIncarnation: + return self.proposed.parent_identity -class CommanderPermission: - """Strikes and releases are the order's commander's to make.""" - actor: ThreadIncarnation - - def require_permitted(self, current: Todo) -> None: - current.require_commander(self.actor) - - -class StallWatcherPermission(CommanderPermission): - """Core's stall watcher makes these on its own declared identity, never the commander's.""" - def require_permitted(self, current: Todo) -> None: - if self.actor != StallWatcherIncarnation.watcher(): - raise TodoConflict(current) +@dataclass(frozen=True) +class StallAssignTodo(StallWatcherPermission, AssignTodo): + """Core's stall watcher gives a released order to an idle unit.""" @dataclass(frozen=True) @@ -2248,7 +2271,31 @@ class TodoStore(TypedStore): root: Path filename: ClassVar[str] = "todos.sqlite3" - rows: ClassVar[tuple[type[TypedTable], ...]] = (Todo, HuntRecord) + rows: ClassVar[tuple[type[TypedTable], ...]] = (Todo, HuntRecord, ParkedThread, StallPassFailed) + + def park(self, name: str) -> None: + with self.writing() as db: + ParkedThread(name).upsert(db) + + def unpark(self, name: str) -> None: + with self.writing() as db: + ParkedThread.delete(db, name=name) + + def parked(self) -> frozenset[str]: + with self.snapshot() as db: + return frozenset(row.name for row in ParkedThread.select(db)) + + def stall_pass_failed(self, failed: StallPassFailed) -> None: + with self.writing() as db: + failed.upsert(db) + + def stall_pass_succeeded(self) -> None: + with self.writing() as db: + StallPassFailed.delete(db) + + def stall_pass_failures(self) -> tuple[StallPassFailed, ...]: + with self.snapshot() as db: + return tuple(StallPassFailed.select(db)) def record_hunt(self, commander: ThreadIncarnation, hunt: Hunt) -> HuntRecord: hunt.require_commander(self, commander) @@ -2501,19 +2548,16 @@ def assign( *, expected_revision: int, owner: Thread, - parent: Thread, + assigner: ThreadIncarnation, + operation: type[AssignTodo], generation: str, goal: GoalRef, deadline: EventTime, order: PendingProblemOrder, traced: FileLine | None = None, ) -> Todo: - identity, parent_identity = _thread(owner), _thread(parent) - proposed = Assignment( - identity.name, - identity.created_at, - parent_identity.name, - parent_identity.created_at, - generation, - ) - return self.apply(todo_id, AssignTodo(expected_revision, proposed, deadline, order, goal, traced)) + identity = _thread(owner) + proposed = Assignment(identity.name, identity.created_at, assigner.name, assigner.wire_birth, + generation) + return self.apply(todo_id, operation(expected_revision, proposed, deadline, order, goal, traced)) + diff --git a/tests/test_stall_watcher_strike.py b/tests/test_stall_watcher_strike.py index e41106c3dd..1faed855ec 100644 --- a/tests/test_stall_watcher_strike.py +++ b/tests/test_stall_watcher_strike.py @@ -1,5 +1,6 @@ -import importlib +import asyncio +from agent_comms import stall_watcher from agent_comms.change_report import FileLine from agent_comms.goal_actions import CreateProblemGoalAction from agent_comms.goals import CountCheckResult, OrderCheck @@ -11,8 +12,6 @@ def test_stall_pass_strikes_releases_and_reassigns_on_the_watcher_identity(commander, commander_script, monkeypatch): comms, caller = commander - from agent_comms.commander import wheel_spin - wheel_spin = importlib.reload(wheel_spin) comms.registry.declare(comms.registry.new_thread("spare", frozenset({'comms'}), str(comms.root), created_at=3.0)) order = PendingProblemOrder(id="order-P1", owner=FileLine(__file__, 1), run=MechanicalOrderRun(), check=OrderCheck(str(comms.root), "printf 1", CountCheckResult(1)), edits=("Repair the owner",), @@ -22,9 +21,9 @@ def test_stall_pass_strikes_releases_and_reassigns_on_the_watcher_identity(comma request="request", deadline=RecordedEventTime(1000), commander="lead", repo="OpenHCSDev/agent-comms", text="Fix P1")).apply(comms, caller) # the one stubbed read: turn-digest activity marks worker and spare as recently active - monkeypatch.setattr(wheel_spin.WheelSpin, "idle", classmethod( + monkeypatch.setattr(stall_watcher.WheelSpin, "idle", classmethod( lambda cls, c, cmd: [c.registry.require("worker"), c.registry.require("spare")])) - wheel_spin.UnownedWheelSpin.after_stall(comms) + stall_watcher.StallWheelSpin.after_stall(comms) todo = comms.goals.todos.get("P1") # read back from the store: the strike keeps its kind and the watcher as its actor assert [type(s) for s in todo.strikes] == [StallStrikeTodo] @@ -33,3 +32,39 @@ def test_stall_pass_strikes_releases_and_reassigns_on_the_watcher_identity(comma assert StallWatcherIncarnation.watcher().name == "stall_watcher" # the same pass gives the order to the idle unit not struck on it, never back to worker assert todo.holder == "spare" and todo.live_strikes("worker") + # Core's watcher made the assignment, while the order stays on its commander's board + assert todo.assignment.parent_identity == StallWatcherIncarnation.watcher() + assert [t.id for t in comms.goals.todos.held_for_commander(comms.registry.require("lead").incarnation)] == ["P1"] + + +def test_a_failed_stall_pass_is_recorded_and_the_next_pass_still_runs(commander, monkeypatch): + comms, _ = commander + passes = [] + + def after_stall(cls, c): + # the second pass runs after the first raised, and sees the first's recorded failure + passes.append([f.error for f in c.goals.todos.stall_pass_failures()]) + if len(passes) == 1: + raise FileNotFoundError("no commander config") + c.goals.todos.stall_pass_succeeded() + raise asyncio.CancelledError + + async def no_wait(seconds): + pass + + monkeypatch.setattr(stall_watcher.StallWheelSpin, "after_stall", classmethod(after_stall)) + monkeypatch.setattr(stall_watcher.asyncio, "sleep", no_wait) + try: + asyncio.run(stall_watcher.StallWheelSpin.watch(comms)) + except asyncio.CancelledError: + pass + assert passes == [[], ["FileNotFoundError: no commander config"]] + assert comms.goals.todos.stall_pass_failures() == () + + +def test_parked_threads_are_core_facts_stored_with_the_todos(commander): + comms, _ = commander + comms.goals.todos.park("worker") + assert comms.goals.todos.parked() == frozenset({"worker"}) + comms.goals.todos.unpark("worker") + assert comms.goals.todos.parked() == frozenset() From 519dd3c66e1ae51d4ffdaaf7f33c5b09c89b3496 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Mon, 19 Oct 2026 17:10:05 -0400 Subject: [PATCH 10/18] P717 keep main's removed-checkout model; delete the competing missing-checkout progress Co-Authored-By: Claude Opus 5.5 --- src/agent_comms/todos.py | 18 ------------------ 1 file changed, 18 deletions(-) diff --git a/src/agent_comms/todos.py b/src/agent_comms/todos.py index 68d3ffe824..5bdf6cf61c 100644 --- a/src/agent_comms/todos.py +++ b/src/agent_comms/todos.py @@ -336,10 +336,6 @@ def readings(self, source: ProgressSource) -> DistanceReadings: ... @abstractmethod def pushes(self, since: float) -> Iterator[tuple[str, float]]: ... - def missing(self, source: ProgressSource) -> Iterator[MissingCheckoutProgress]: - """Only a missing checkout is progress of its own kind.""" - return iter(()) - class MissingCheckout(ProgressCheckout): """A retired checkout contributes no local history, not a failed Git read.""" @@ -353,10 +349,6 @@ def readings(self, source: ProgressSource) -> DistanceReadings: def pushes(self, since: float) -> Iterator[tuple[str, float]]: return iter(()) - def missing(self, source: ProgressSource) -> Iterator[MissingCheckoutProgress]: - yield MissingCheckoutProgress( - at=max((source.since, *(pr.updated_at for pr in source.requests("open"))))) - class PresentCheckout(ProgressCheckout): def git(self, *args: str) -> str: @@ -569,16 +561,6 @@ def distance_readings(self) -> DistanceReadings: return self.readings -@dataclass(frozen=True, kw_only=True) -class MissingCheckoutProgress(Moving, FoundProgress): - precedence: ClassVar[int] = 2 - state: ClassVar[str] = "ordered checkout is missing" - - @classmethod - def found(cls, source: ProgressSource) -> Iterator[Self]: - return source.checkout.missing(source) - - @dataclass(frozen=True) class NoCommitProgress(Stalling, OrderProgress): def missing_merges(self, source: ProgressSource) -> tuple[MergedProgress, ...]: From 22f74ddaf63e887bcf34bf5c4fc81c701a7a0e89 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Mon, 19 Oct 2026 17:18:41 -0400 Subject: [PATCH 11/18] P717 supervise each stall pass as its own task; migrate assign callers to the assigner Co-Authored-By: Claude Opus 5.5 --- src/agent_comms/goal_actions.py | 15 +++++++++++---- src/agent_comms/stall_watcher.py | 18 +++++++++++++----- src/agent_comms/todos.py | 7 ++++--- tests/test_commander_run_rules.py | 6 +++--- tests/test_todos.py | 4 ++-- 5 files changed, 33 insertions(+), 17 deletions(-) diff --git a/src/agent_comms/goal_actions.py b/src/agent_comms/goal_actions.py index c33c6da20f..3b82db901b 100644 --- a/src/agent_comms/goal_actions.py +++ b/src/agent_comms/goal_actions.py @@ -27,21 +27,28 @@ 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 .threads import Thread from .thread_identity import StallWatcherIncarnation, ThreadIncarnation -from .todos import (AssignTodo, GoalRef, PendingProblemOrder, SendBackVerdict, StallAssignTodo, Todo, - TodoOperation) +from .threads import Thread +from .todos import ( + AssignTodo, + GoalRef, + PendingProblemOrder, + SendBackVerdict, + StallAssignTodo, + Todo, + TodoOperation, +) if TYPE_CHECKING: from .goal_management import Goals diff --git a/src/agent_comms/stall_watcher.py b/src/agent_comms/stall_watcher.py index d655348e23..d74280008b 100644 --- a/src/agent_comms/stall_watcher.py +++ b/src/agent_comms/stall_watcher.py @@ -128,11 +128,17 @@ async def watch(cls, comms: Comms) -> None: """The running Core service makes one stall pass a minute; a pass that raises is recorded for the board and the next pass still runs.""" while True: + stall_pass = asyncio.create_task( + Coordination.run_worker(lambda: cls.after_stall(comms)) + ) try: - await Coordination.run_worker(lambda: cls.after_stall(comms)) - except Exception as error: # noqa: BLE001 - recorded on the board, never swallowed + await asyncio.wait((stall_pass,)) + finally: + stall_pass.cancel() + if (failure := stall_pass.exception()) is not None: comms.goals.todos.stall_pass_failed( - StallPassFailed(time.time(), f"{type(error).__name__}: {error}")) + StallPassFailed(time.time(), f"{type(failure).__name__}: {failure}") + ) await asyncio.sleep(60) @@ -148,5 +154,7 @@ def violations(cls, comms: Comms, commander: str) -> tuple[Self, ...]: def line(self) -> str: at = time.strftime("%H:%M", time.localtime(self.failed.at)) - return (f"STALL WATCHER FAILED at {at}: {self.failed.error} -> fix the cause; the next" - " pass runs within a minute") + return ( + f"STALL WATCHER FAILED at {at}: {self.failed.error} -> fix the cause; the next" + " pass runs within a minute" + ) diff --git a/src/agent_comms/todos.py b/src/agent_comms/todos.py index d72fad8a93..d8ba38fb60 100644 --- a/src/agent_comms/todos.py +++ b/src/agent_comms/todos.py @@ -2605,7 +2605,8 @@ def assign( traced: FileLine | None = None, ) -> Todo: identity = _thread(owner) - proposed = Assignment(identity.name, identity.created_at, assigner.name, assigner.wire_birth, - generation) - return self.apply(todo_id, operation(expected_revision, proposed, deadline, order, goal, traced)) + proposed = Assignment( + identity.name, identity.created_at, assigner.name, assigner.wire_birth, generation) + return self.apply( + todo_id, operation(expected_revision, proposed, deadline, order, goal, traced)) diff --git a/tests/test_commander_run_rules.py b/tests/test_commander_run_rules.py index 12475225d0..ccfd7739fd 100644 --- a/tests/test_commander_run_rules.py +++ b/tests/test_commander_run_rules.py @@ -6,7 +6,7 @@ from agent_comms.change_report import FileLine from agent_comms.field_codec import FieldCodec from agent_comms.recorded_time import RecordedEventTime -from agent_comms.todos import ApplicationOrderRun, GoalRef, MechanicalOrderRun, Todo, TodoStore +from agent_comms.todos import ApplicationOrderRun, AssignTodo, GoalRef, MechanicalOrderRun, Todo, TodoStore from test_todos import order, thread @@ -31,11 +31,11 @@ def test_clean_transfers_keep_history_and_third_owner_needs_a_real_line(tmp_path creator=lead, owner=lead, generation='one', goal=GoalRef('lead', 1, 'one'), deadline=RecordedEventTime(1000), order=order(tmp_path)) second = store.assign('P1', expected_revision=first.revision, - owner=thread('second', tmp_path, 2), parent=lead, generation='two', goal=GoalRef('second', 2, 'two'), + owner=thread('second', tmp_path, 2), assigner=lead.incarnation, operation=AssignTodo, generation='two', goal=GoalRef('second', 2, 'two'), deadline=RecordedEventTime(1000), order=order(tmp_path)) assert not second.strikes kwargs = dict(expected_revision=second.revision, owner=thread('third', tmp_path, 3), - parent=lead, generation='three', goal=GoalRef('third', 3, 'three'), deadline=RecordedEventTime(1000), order=order(tmp_path)) + assigner=lead.incarnation, operation=AssignTodo, generation='three', goal=GoalRef('third', 3, 'three'), deadline=RecordedEventTime(1000), order=order(tmp_path)) with pytest.raises(ValueError, match='--traced PATH:LINE'): store.assign('P1', **kwargs) with pytest.raises(ValueError, match='does not exist'): diff --git a/tests/test_todos.py b/tests/test_todos.py index 1fe58fe3b9..2e9aaf1cad 100644 --- a/tests/test_todos.py +++ b/tests/test_todos.py @@ -18,7 +18,7 @@ from agent_comms.registration import Registration from agent_comms.threads import Thread from agent_comms.todos import ( - Assignment, ChangeTodoState, DoneTodoState, GoalRef, MechanicalOrderRun, PendingProblemOrder, + AssignTodo, Assignment, ChangeTodoState, DoneTodoState, GoalRef, MechanicalOrderRun, PendingProblemOrder, BlockedTodoState, OperationDelivery, Todo, TodoConflict, TodoError, TodoState, TodoStore, ) @@ -43,7 +43,7 @@ def _assign_process(root: str, name: str, created: float, gate) -> tuple[str, st try: row = store.assign( "todo-1", expected_revision=1, - owner=thread(name, Path(root), created), parent=thread("lead", Path(root), 1.0), + owner=thread(name, Path(root), created), assigner=thread("lead", Path(root), 1.0).incarnation, operation=AssignTodo, generation=f"generation-{name}", goal=GoalRef("lead", 1.0, "goal-1"), deadline=RecordedEventTime(1000.0), order=order(Path(root)), ) From e9e889a2e98ec9410af0ce659fba4bfab51950d0 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Mon, 19 Oct 2026 17:20:27 -0400 Subject: [PATCH 12/18] P717 record a stall pass's outcome by its type The watcher's recorded failure state, the Core parked set with its CLI and the watcher's own assignment action are new behavior the review asked for. Src-Budget: 23 Co-Authored-By: Claude Opus 5.5 --- src/agent_comms/stall_watcher.py | 17 +++++++---------- tests/test_stall_watcher_strike.py | 8 ++++---- 2 files changed, 11 insertions(+), 14 deletions(-) diff --git a/src/agent_comms/stall_watcher.py b/src/agent_comms/stall_watcher.py index d74280008b..1fb4796ff3 100644 --- a/src/agent_comms/stall_watcher.py +++ b/src/agent_comms/stall_watcher.py @@ -128,17 +128,14 @@ async def watch(cls, comms: Comms) -> None: """The running Core service makes one stall pass a minute; a pass that raises is recorded for the board and the next pass still runs.""" while True: - stall_pass = asyncio.create_task( - Coordination.run_worker(lambda: cls.after_stall(comms)) + (outcome,) = await asyncio.gather( + Coordination.run_worker(lambda: cls.after_stall(comms)), return_exceptions=True ) - try: - await asyncio.wait((stall_pass,)) - finally: - stall_pass.cancel() - if (failure := stall_pass.exception()) is not None: - comms.goals.todos.stall_pass_failed( - StallPassFailed(time.time(), f"{type(failure).__name__}: {failure}") - ) + match outcome: + case Exception() as failure: + comms.goals.todos.stall_pass_failed( + StallPassFailed(time.time(), f"{type(failure).__name__}: {failure}") + ) await asyncio.sleep(60) diff --git a/tests/test_stall_watcher_strike.py b/tests/test_stall_watcher_strike.py index 1faed855ec..1292abba8a 100644 --- a/tests/test_stall_watcher_strike.py +++ b/tests/test_stall_watcher_strike.py @@ -47,13 +47,13 @@ def after_stall(cls, c): if len(passes) == 1: raise FileNotFoundError("no commander config") c.goals.todos.stall_pass_succeeded() - raise asyncio.CancelledError - async def no_wait(seconds): - pass + async def wait(seconds): # the watcher stops after its second pass + if len(passes) == 2: + raise asyncio.CancelledError monkeypatch.setattr(stall_watcher.StallWheelSpin, "after_stall", classmethod(after_stall)) - monkeypatch.setattr(stall_watcher.asyncio, "sleep", no_wait) + monkeypatch.setattr(stall_watcher.asyncio, "sleep", wait) try: asyncio.run(stall_watcher.StallWheelSpin.watch(comms)) except asyncio.CancelledError: From d58a4902f3de14884e115057ab2492b19959f79c Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Mon, 19 Oct 2026 18:53:47 -0400 Subject: [PATCH 13/18] Keep unowned orders eligible for later watcher passes --- src/agent_comms/todos.py | 4 +--- tests/test_stall_watcher_strike.py | 17 +++++++++++++++++ 2 files changed, 18 insertions(+), 3 deletions(-) diff --git a/src/agent_comms/todos.py b/src/agent_comms/todos.py index 233427002a..12fbe45295 100644 --- a/src/agent_comms/todos.py +++ b/src/agent_comms/todos.py @@ -2420,7 +2420,6 @@ def advance_stalled(self, storage: StorageRule) -> tuple[Todo, ...]: """Observe outside SQLite, then atomically strike and release the same assignment.""" held = tuple(todo for todo in self._open("assignment IS NOT NULL", ()) if todo.pending) pulls = PullRequests(held) - changed = [] for todo in held: observed_at = time.time() source = ProgressSource.of(todo, storage, pulls) @@ -2432,11 +2431,10 @@ def advance_stalled(self, storage: StorageRule) -> tuple[Todo, ...]: result = progress.advance(current, source.since, observed_at) if result != current: result.upsert(db) - changed.append(result) except TodoConflict: # A concurrent order change owns the row; observe its new assignment next pass. continue - return tuple(changed) + return self._open("assignment IS NULL", ()) def board_disagreements( self, commander: ThreadIncarnation, storage: StorageRule, diff --git a/tests/test_stall_watcher_strike.py b/tests/test_stall_watcher_strike.py index 8b86a407d6..090d35a59d 100644 --- a/tests/test_stall_watcher_strike.py +++ b/tests/test_stall_watcher_strike.py @@ -1,5 +1,8 @@ import asyncio +import json +import shutil +from agent_comms.comms import Comms from agent_comms.commander import standing from agent_comms.change_report import FileLine from agent_comms.goal_actions import CreateProblemGoalAction @@ -20,6 +23,10 @@ def test_stall_pass_strikes_releases_and_reassigns_on_the_watcher_identity(comma CommsTodoCreateTool(owner="worker", order=CreateProblemGoalAction(problem="P1", order=order, request="request", deadline=RecordedEventTime(1000), commander="lead", repo="OpenHCSDev/agent-comms", text="Fix P1")).apply(comms, caller) + original = comms.root + copied = original.with_name("copied-wire") + shutil.copytree(original, copied) + comms = Comms(copied) # the one stubbed read: turn-digest activity marks worker and spare as recently active monkeypatch.setattr(standing.WheelSpin, "idle", classmethod( lambda cls, c, cmd: [c.registry.require("worker"), c.registry.require("spare")])) @@ -35,6 +42,16 @@ def test_stall_pass_strikes_releases_and_reassigns_on_the_watcher_identity(comma # Core's watcher made the assignment, while the order stays on its commander's board assert todo.assignment.parent_identity == StallWatcherIncarnation.watcher() assert [t.id for t in comms.goals.todos.held_for_commander(comms.registry.require("lead").incarnation)] == ["P1"] + source = Comms(original).goals.todos.get("P1") + assert source.holder == "worker" and not source.strikes and not source.released + (copied.parent / "copied-root-result.json").write_text(json.dumps({ + "original": str(original), "copy": str(copied), "id": todo.id, + "holder": todo.holder, "revision": todo.revision, + "strike_actor": todo.strikes[0].actor.to_wire(), + "release_actor": todo.released[0].actor.to_wire(), + "assigner": todo.held().parent_identity.to_wire(), + "original_holder": source.holder, "original_strikes": len(source.strikes), + "qualification": "Copied fixture root; recent native activity is controlled, no commander invocation."}, indent=2)) def test_a_failed_stall_pass_is_recorded_and_the_next_pass_still_runs(commander, monkeypatch): From f59dc285d6122dee1a586f9d296e2625ac16f22f Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Mon, 19 Oct 2026 19:22:24 -0400 Subject: [PATCH 14/18] Scope stall passes to active watcher releases and release idle route locks --- src/agent_comms/cli_commands.py | 20 ----- src/agent_comms/commander/config.py | 1 + src/agent_comms/commander/standing.py | 49 +++++------- src/agent_comms/connector.py | 15 ++-- src/agent_comms/goal_actions.py | 8 +- src/agent_comms/todos.py | 55 +++---------- tests/test_stall_watcher_strike.py | 109 +++++++++++++++++--------- tests/test_standing_rules.py | 11 ++- 8 files changed, 118 insertions(+), 150 deletions(-) diff --git a/src/agent_comms/cli_commands.py b/src/agent_comms/cli_commands.py index 8df0c0c925..c69ae0b9d8 100644 --- a/src/agent_comms/cli_commands.py +++ b/src/agent_comms/cli_commands.py @@ -645,11 +645,6 @@ class RegisteredResult(CliResult): registered: str -@dataclass(frozen=True) -class ParkedResult(CliResult): - parked: tuple[str, ...] - - @dataclass(frozen=True) class HeartbeatResult(CliResult): heartbeat: str @@ -1483,21 +1478,6 @@ def apply(self, ctx: Comms) -> CliResult: for harness in Harness.present(home)}) -@dataclass(frozen=True, kw_only=True) -class ParkCliCommand(CliCommand): - help = "Keep Core's stall watcher from giving orders to a thread, or allow it again with --off" - name: str = option("--name") - off: bool = option("--off", default=False) - - def apply(self, ctx: Comms) -> CliResult: - todos = ctx.goals.todos - if self.off: - todos.unpark(self.name) - else: - todos.park(ctx.registry.require(self.name).name) - return ParkedResult(tuple(sorted(todos.parked()))) - - @dataclass(frozen=True, kw_only=True) class HeartbeatCliCommand(CliCommand): help = "Mark thread running" diff --git a/src/agent_comms/commander/config.py b/src/agent_comms/commander/config.py index 92851d41e7..df6c020045 100644 --- a/src/agent_comms/commander/config.py +++ b/src/agent_comms/commander/config.py @@ -72,6 +72,7 @@ class CommanderConfig(CommanderRole, BaseModel): worktrees: Path scratch: Path disk: Path + parked: frozenset[str] = frozenset() @staticmethod def path(home: Path | None = None) -> Path: diff --git a/src/agent_comms/commander/standing.py b/src/agent_comms/commander/standing.py index 084dbf5805..0f4b0c167b 100644 --- a/src/agent_comms/commander/standing.py +++ b/src/agent_comms/commander/standing.py @@ -3,12 +3,14 @@ from __future__ import annotations import asyncio +import logging import subprocess import time -from collections.abc import Iterable, Iterator +from collections.abc import Callable, Iterable, Iterator +from contextlib import AbstractContextManager from dataclasses import dataclass from pathlib import Path -from typing import ClassVar, Self +from typing import ClassVar from ..comms import Comms from ..coordinator import Coordination @@ -23,7 +25,7 @@ from ..thread_identity import StallWatcherIncarnation, ThreadIncarnation from ..threads import Thread from ..todos import ( - FlatTrend, OrderProgress, PullRequests, StallPassFailed, StrikeTodo, Todo, TodoConflict, + FlatTrend, OrderProgress, PullRequests, StrikeTodo, Todo, TodoConflict, ) from ..tools import CommsUnjudgedTool from ..turn_digests import TurnDigestRead, TurnsSince @@ -81,11 +83,11 @@ 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 | comms.goals.todos.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).native()), + and TurnDigestRead(comms.root, thread, bound).read().turns), key=lambda thread: thread.name) def assign(self, comms: Comms, commander: str, unit: Thread, @@ -182,35 +184,22 @@ def after_stall(cls, comms: Comms) -> None: StallWatcherIncarnation.watcher()) except TodoConflict: continue # A concurrent order change owns the row. - comms.goals.todos.stall_pass_succeeded() @classmethod - async def watch(cls, comms: Comms) -> None: - """Run a pass each minute, recording failures without abandoning the watcher.""" - while True: - (outcome,) = await asyncio.gather( - Coordination.run_worker(lambda: cls.after_stall(comms)), return_exceptions=True) - match outcome: - case Exception() as failure: - comms.goals.todos.stall_pass_failed( - StallPassFailed(time.time(), f"{type(failure).__name__}: {failure}")) - await asyncio.sleep(60) - - -@dataclass(frozen=True) -class StallWatcherFailed(UndatedStandingRule): - """A stall watcher pass raised and no pass has succeeded since.""" - - failed: StallPassFailed + 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 - def violations(cls, comms: Comms, commander: str) -> tuple[Self, ...]: - return tuple(map(cls, comms.goals.todos.stall_pass_failures())) - - def line(self) -> str: - at = time.strftime("%H:%M", time.localtime(self.failed.at)) - return (f"STALL WATCHER FAILED at {at}: {self.failed.error} -> fix the cause; the next" - " pass runs within a minute") + async def watch(cls, opened_wire: Callable[[], AbstractContextManager[Comms]]) -> None: + """Run a pass each minute; failures go to the existing service log.""" + while True: + try: + await Coordination.run_worker(lambda: cls.watch_pass(opened_wire)) + except Exception: + logging.exception("Stall watcher pass failed") + await asyncio.sleep(60) @dataclass(frozen=True) diff --git a/src/agent_comms/connector.py b/src/agent_comms/connector.py index 5445fda3fb..7b37fdecc9 100644 --- a/src/agent_comms/connector.py +++ b/src/agent_comms/connector.py @@ -376,14 +376,13 @@ def application(self) -> web.Application: async def watch_orders(self, app: web.Application) -> AsyncIterator[None]: """The running Core service advances stalled orders without a commander call.""" - with self.opened_wire() as comms: - task = asyncio.create_task(UnownedWheelSpin.watch(comms)) - try: - yield - finally: - task.cancel() - with suppress(asyncio.CancelledError): - await task + 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( diff --git a/src/agent_comms/goal_actions.py b/src/agent_comms/goal_actions.py index e09c04970b..c011d6f757 100644 --- a/src/agent_comms/goal_actions.py +++ b/src/agent_comms/goal_actions.py @@ -40,13 +40,7 @@ from .recorded_time import EventTime from .thread_identity import ThreadIncarnation from .threads import Thread -from .todos import ( - GoalRef, - PendingProblemOrder, - SendBackVerdict, - Todo, - TodoOperation, -) +from .todos import GoalRef, PendingProblemOrder, SendBackVerdict, Todo, TodoOperation if TYPE_CHECKING: from .goal_management import Goals diff --git a/src/agent_comms/todos.py b/src/agent_comms/todos.py index 2c80bbe531..1e6b47dc45 100644 --- a/src/agent_comms/todos.py +++ b/src/agent_comms/todos.py @@ -1550,19 +1550,6 @@ class HuntRecord(TypedTable): hunt: Hunt -@dataclass(frozen=True) -class ParkedThread(TypedTable): - """A thread Core's stall watcher never gives an order to.""" - name: str = field(metadata={"sql": Column(primary_key=True)}) - - -@dataclass(frozen=True) -class StallPassFailed(TypedTable): - """A stall watcher pass that raised; the board shows it until a later pass succeeds.""" - at: float = field(metadata={"sql": Column(primary_key=True)}) - error: str - - class TodoError(ToolRefusedError): """A todo request is invalid or its persisted state cannot be trusted.""" @@ -2322,31 +2309,7 @@ class TodoStore(TypedStore): root: Path filename: ClassVar[str] = "todos.sqlite3" - rows: ClassVar[tuple[type[TypedTable], ...]] = (Todo, HuntRecord, ParkedThread, StallPassFailed) - - def park(self, name: str) -> None: - with self.writing() as db: - ParkedThread(name).upsert(db) - - def unpark(self, name: str) -> None: - with self.writing() as db: - ParkedThread.delete(db, name=name) - - def parked(self) -> frozenset[str]: - with self.snapshot() as db: - return frozenset(row.name for row in ParkedThread.select(db)) - - def stall_pass_failed(self, failed: StallPassFailed) -> None: - with self.writing() as db: - failed.upsert(db) - - def stall_pass_succeeded(self) -> None: - with self.writing() as db: - StallPassFailed.delete(db) - - def stall_pass_failures(self) -> tuple[StallPassFailed, ...]: - with self.snapshot() as db: - return tuple(StallPassFailed.select(db)) + rows: ClassVar[tuple[type[TypedTable], ...]] = (Todo, HuntRecord) def record_hunt(self, commander: ThreadIncarnation, hunt: Hunt) -> HuntRecord: hunt.require_commander(self, commander) @@ -2442,8 +2405,9 @@ def for_goal(self, owner: ThreadIncarnation, goal_id: str) -> tuple[Todo, ...]: "AND json_extract(assignment,'$.generation')=?", (owner.name, owner.created_at, goal_id)) def advance_stalled(self, storage: StorageRule) -> tuple[Todo, ...]: - """Observe outside SQLite, then atomically strike and release the same assignment.""" - held = tuple(todo for todo in self._open("assignment IS NOT NULL", ()) if todo.pending) + """Advance active orders; retry only rows released by this watcher, never other backlog.""" + held = tuple(todo for todo in self._open( + "assignment IS NOT NULL", (), OpenTodoState) if todo.pending) pulls = PullRequests(held) for todo in held: observed_at = time.time() @@ -2459,7 +2423,11 @@ def advance_stalled(self, storage: StorageRule) -> tuple[Todo, ...]: except TodoConflict: # A concurrent order change owns the row; observe its new assignment next pass. continue - return self._open("assignment IS NULL", ()) + watcher = StallWatcherIncarnation.watcher() + return self._open( + "assignment IS NULL AND json_extract(released,'$[#-1].actor.name')=? " + "AND json_extract(released,'$[#-1].actor.created_at')=?", + (watcher.name, watcher.wire_birth), OpenTodoState) def reconcile_merges(self, commander: ThreadIncarnation, repos: Sequence[Repo], storage: StorageRule) -> None: @@ -2528,8 +2496,9 @@ def units(self, commander: ThreadIncarnation) -> frozenset[str]: for assignment in (todo.assignment, *(release.previous for release in todo.released)) if assignment) - def _open(self, where: str, parameters: tuple) -> tuple[Todo, ...]: - states = TodoState.open_names() + def _open(self, where: str, parameters: tuple, + state: type[TodoState] = TodoState) -> tuple[Todo, ...]: + states = state.open_names() selected = ','.join('?' for state in states) return self._select(f"state IN ({selected}) AND ({where})", (*states, *parameters)) diff --git a/tests/test_stall_watcher_strike.py b/tests/test_stall_watcher_strike.py index 090d35a59d..20e14451fd 100644 --- a/tests/test_stall_watcher_strike.py +++ b/tests/test_stall_watcher_strike.py @@ -1,46 +1,35 @@ import asyncio import json import shutil +from contextlib import contextmanager +from pathlib import Path from agent_comms.comms import Comms from agent_comms.commander import standing -from agent_comms.change_report import FileLine -from agent_comms.goal_actions import CreateProblemGoalAction -from agent_comms.goals import CountCheckResult, OrderCheck -from agent_comms.recorded_time import RecordedEventTime +from agent_comms.commander.config import CommanderConfig from agent_comms.thread_identity import StallWatcherIncarnation -from agent_comms.todos import MechanicalOrderRun, PendingProblemOrder, ReleaseTodo, StrikeTodo -from agent_comms.tools import CommsTodoCreateTool +from agent_comms.todos import BlockedTodoState, ChangeTodoState, ReleaseTodo, StrikeReason, StrikeTodo +from test_commander_problem_orders import create +from test_standing_rules import working -def test_stall_pass_strikes_releases_and_reassigns_on_the_watcher_identity(commander, commander_script, monkeypatch): - comms, caller = commander +def test_stall_pass_strikes_releases_and_reassigns_on_the_watcher_identity(commander, commander_script): + comms, _ = commander comms.registry.declare(comms.registry.new_thread("spare", frozenset({'comms'}), str(comms.root), created_at=3.0)) - order = PendingProblemOrder(id="order-P1", owner=FileLine(__file__, 1), run=MechanicalOrderRun(), - check=OrderCheck(str(comms.root), "printf 1", CountCheckResult(1)), edits=("Repair the owner",), - banned=("No second store",), report="Report the result", - at=RecordedEventTime(100), first_call_deadline=RecordedEventTime(220)) - CommsTodoCreateTool(owner="worker", order=CreateProblemGoalAction(problem="P1", order=order, - request="request", deadline=RecordedEventTime(1000), commander="lead", - repo="OpenHCSDev/agent-comms", text="Fix P1")).apply(comms, caller) + create(commander, 'P1') + working(comms, 'worker', 2.0) + working(comms, 'spare', 3.0) original = comms.root copied = original.with_name("copied-wire") shutil.copytree(original, copied) comms = Comms(copied) - # the one stubbed read: turn-digest activity marks worker and spare as recently active - monkeypatch.setattr(standing.WheelSpin, "idle", classmethod( - lambda cls, c, cmd: [c.registry.require("worker"), c.registry.require("spare")])) standing.UnownedWheelSpin.after_stall(comms) todo = comms.goals.todos.get("P1") - # read back from the store: the strike keeps its kind and the watcher as its actor assert [type(s) for s in todo.strikes] == [StrikeTodo] assert [type(r) for r in todo.released] == [ReleaseTodo] assert todo.strikes[0].actor == todo.released[0].actor == StallWatcherIncarnation.watcher() - assert StallWatcherIncarnation.watcher().name == "stall_watcher" - # the same pass gives the order to the idle unit not struck on it, never back to worker assert todo.holder == "spare" and todo.live_strikes("worker") - # Core's watcher made the assignment, while the order stays on its commander's board - assert todo.assignment.parent_identity == StallWatcherIncarnation.watcher() + assert todo.held().parent_identity == StallWatcherIncarnation.watcher() assert [t.id for t in comms.goals.todos.held_for_commander(comms.registry.require("lead").incarnation)] == ["P1"] source = Comms(original).goals.todos.get("P1") assert source.holder == "worker" and not source.strikes and not source.released @@ -51,37 +40,81 @@ def test_stall_pass_strikes_releases_and_reassigns_on_the_watcher_identity(comma "release_actor": todo.released[0].actor.to_wire(), "assigner": todo.held().parent_identity.to_wire(), "original_holder": source.holder, "original_strikes": len(source.strikes), - "qualification": "Copied fixture root; recent native activity is controlled, no commander invocation."}, indent=2)) + "qualification": "Copied fixture root and controlled saved native records; actual history reader, no commander invocation."}, indent=2)) -def test_a_failed_stall_pass_is_recorded_and_the_next_pass_still_runs(commander, monkeypatch): +def test_a_failed_stall_pass_logs_and_releases_the_root_before_the_next_pass(commander, monkeypatch, caplog): comms, _ = commander - passes = [] + passes, borrowed = [], [] + + @contextmanager + def opened_wire(): + borrowed.append(comms) + try: + yield comms + finally: + borrowed.remove(comms) def after_stall(cls, c): - # the second pass runs after the first raised, and sees the first's recorded failure - passes.append([f.error for f in c.goals.todos.stall_pass_failures()]) + assert borrowed == [c] + passes.append(c) if len(passes) == 1: raise FileNotFoundError("no commander config") - c.goals.todos.stall_pass_succeeded() - async def wait(seconds): # the watcher stops after its second pass + async def wait(seconds): + assert not borrowed if len(passes) == 2: raise asyncio.CancelledError monkeypatch.setattr(standing.UnownedWheelSpin, "after_stall", classmethod(after_stall)) monkeypatch.setattr(standing.asyncio, "sleep", wait) try: - asyncio.run(standing.UnownedWheelSpin.watch(comms)) + asyncio.run(standing.UnownedWheelSpin.watch(opened_wire)) except asyncio.CancelledError: pass - assert passes == [[], ["FileNotFoundError: no commander config"]] - assert comms.goals.todos.stall_pass_failures() == () + assert len(passes) == 2 and not borrowed + assert "Stall watcher pass failed" in caplog.text + assert "FileNotFoundError: no commander config" in caplog.text + + +def test_parked_units_come_from_the_existing_commander_configuration(commander, commander_script): + comms, _ = commander + config = CommanderConfig.installed().model_copy(update={"parked": frozenset({"worker"})}) + config.write(Path.home()) + CommanderConfig.installed.cache_clear() + working(comms, 'worker', 2.0) + assert standing.WheelSpin.idle(comms, "lead") == [] + + +def test_blocked_held_orders_are_not_struck_or_released(commander): + comms, _ = commander + todo = create(commander, 'P1') + blocked = comms.goals.todos.apply('P1', ChangeTodoState( + todo.revision, BlockedTodoState, comms.registry.require('lead').incarnation, None)) + standing.UnownedWheelSpin.after_stall(comms) + assert comms.goals.todos.get('P1') == blocked + + +def test_commander_released_backlog_is_not_automatically_reassigned(commander): + comms, _ = commander + todo = create(commander, 'P1') + released = comms.goals.todos.apply('P1', ReleaseTodo( + todo.revision, comms.registry.require('lead').incarnation, todo.held(), + StrikeReason('Commander intentionally froze this unfinished order without authorizing automatic reassignment.'))) + standing.UnownedWheelSpin.after_stall(comms) + assert comms.goals.todos.get('P1') == released -def test_parked_threads_are_core_facts_stored_with_the_todos(commander): +def test_watcher_release_waits_for_a_spare_without_striking_twice(commander, commander_script): comms, _ = commander - comms.goals.todos.park("worker") - assert comms.goals.todos.parked() == frozenset({"worker"}) - comms.goals.todos.unpark("worker") - assert comms.goals.todos.parked() == frozenset() + create(commander, 'P1') + standing.UnownedWheelSpin.after_stall(comms) + waiting = comms.goals.todos.get('P1') + assert waiting.holder is None and len(waiting.strikes) == len(waiting.released) == 1 + comms.registry.declare(comms.registry.new_thread("spare", frozenset({'comms'}), str(comms.root), created_at=3.0)) + working(comms, 'worker', 2.0) + working(comms, 'spare', 3.0) + standing.UnownedWheelSpin.after_stall(comms) + assigned = comms.goals.todos.get('P1') + assert assigned.holder == 'spare' + assert len(assigned.strikes) == len(assigned.released) == 1 diff --git a/tests/test_standing_rules.py b/tests/test_standing_rules.py index 18162603fa..1c1288bfbe 100644 --- a/tests/test_standing_rules.py +++ b/tests/test_standing_rules.py @@ -6,6 +6,7 @@ import sys import time from dataclasses import replace +from pathlib import Path import pytest @@ -150,7 +151,7 @@ def test_idle_reads_external_execution_and_requires_an_actual_recent_turn(standi comms.registry.declare(comms.registry.new_thread( name, frozenset(), str(tmp_path), execution=ExternalThreadExecution)) comms.transcripts.external("external-working").finish( - turn_id="current", item_id="answer", events=(AssistantTranscript("working", position="answer"),)) + turn_id="current", item_id="answer", events=(AssistantTranscript("working", position="answer", timestamp=time.time()),)) names = [thread.name for thread in module.WheelSpin.idle(comms, "lead")] assert "external-working" in names assert "external-empty" not in names and "worker" not in names @@ -229,10 +230,11 @@ def progress(): def test_unjudged_follows_the_unjudged_edits(commander, standing): + from agent_comms.commander import judge comms, state, module = standing assert lines(comms, module.Unjudged) == [] create(commander, 'P1') - native_judge_edits(commander, module.judge, [dict(id=f'e{n}', at=200., path='a.py', + native_judge_edits(commander, judge, [dict(id=f'e{n}', at=200., path='a.py', hunks=[dict(old='before', new='after')]) for n in range(70)]) assert lines(comms, module.Unjudged) == ["UNJUDGED 70 edits -> judge or fork a judge slice"] @@ -372,8 +374,9 @@ def git(*args): git("checkout", "-q", "main") git("merge", "-q", "--no-ff", "-m", f"Merge {branch}", branch) git("push", "-q", "origin", "main") - monkeypatch.setattr(module, "CONFIG", module.CONFIG.model_copy(update={"repos": ( - Repo(checkout=checkout, github="fleet/growth"),)})) + CommanderConfig.installed().model_copy(update={"repos": ( + Repo(checkout=checkout, github="fleet/growth"),)}).write(Path.home()) + CommanderConfig.installed.cache_clear() big = git("rev-parse", "main~1")[:9] assert lines(comms, module.MergeGrowth) == [ f"GROWTH fleet/growth {big} src +41/-0 (net +41) tests +0/-0: Merge big " From 17ee967f322731f4ca66fa03dda8038304c0cbc0 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Mon, 19 Oct 2026 19:30:51 -0400 Subject: [PATCH 15/18] Record stall consequences only with a successful replacement assignment --- src/agent_comms/commander/standing.py | 46 +++++------- src/agent_comms/thread_identity.py | 6 ++ src/agent_comms/todos.py | 60 +++++---------- tests/test_stall_watcher_strike.py | 104 +++++++++++++++++++++++--- 4 files changed, 138 insertions(+), 78 deletions(-) diff --git a/src/agent_comms/commander/standing.py b/src/agent_comms/commander/standing.py index 0f4b0c167b..6f56bb955d 100644 --- a/src/agent_comms/commander/standing.py +++ b/src/agent_comms/commander/standing.py @@ -25,7 +25,7 @@ from ..thread_identity import StallWatcherIncarnation, ThreadIncarnation from ..threads import Thread from ..todos import ( - FlatTrend, OrderProgress, PullRequests, StrikeTodo, Todo, TodoConflict, + FlatTrend, OrderProgress, PullRequests, Todo, ) from ..tools import CommsUnjudgedTool from ..turn_digests import TurnDigestRead, TurnsSince @@ -93,7 +93,7 @@ def idle(cls, comms: Comms, commander: str) -> list[Thread]: 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, @@ -129,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, comms.registry.require(commander).incarnation) + self.assign(comms, commander, unit, StallWatcherIncarnation.watcher()) @dataclass(frozen=True) @@ -148,15 +141,16 @@ class UnownedWheelSpin(WheelSpin, UndatedStandingRule): @classmethod def violations(cls, comms: Comms, commander: str) -> tuple[UnownedWheelSpin, ...]: - return cls.paired(comms, commander, - comms.goals.todos.for_commander(comms.registry.require(commander).incarnation)) + 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 unowned pending row with the first idle unit not struck on it.""" + """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 and not todo.holder): + 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) @@ -174,16 +168,13 @@ def enforce(self, comms: Comms, commander: str) -> None: @classmethod def after_stall(cls, comms: Comms) -> None: - """Strike and release stalled orders, then assign without a commander call.""" - released = comms.goals.todos.advance_stalled(comms.registry.storage_rule) - for commander in {todo.creator_identity.name for todo in released}: - mine = (todo for todo in released if todo.creator_identity.name == commander) + """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): - try: - rule.assign(comms, commander, comms.registry.require(rule.unit), - StallWatcherIncarnation.watcher()) - except TodoConflict: - continue # A concurrent order change owns the row. + rule.assign(comms, commander, comms.registry.require(rule.unit), + StallWatcherIncarnation.watcher()) @classmethod def watch_pass(cls, opened_wire: Callable[[], AbstractContextManager[Comms]]) -> None: @@ -195,10 +186,11 @@ def watch_pass(cls, opened_wire: Callable[[], AbstractContextManager[Comms]]) -> async def watch(cls, opened_wire: Callable[[], AbstractContextManager[Comms]]) -> None: """Run a pass each minute; failures go to the existing service log.""" while True: - try: - await Coordination.run_worker(lambda: cls.watch_pass(opened_wire)) - except Exception: - logging.exception("Stall watcher pass failed") + (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) diff --git a/src/agent_comms/thread_identity.py b/src/agent_comms/thread_identity.py index 63d17ddd84..6b20289488 100644 --- a/src/agent_comms/thread_identity.py +++ b/src/agent_comms/thread_identity.py @@ -95,6 +95,9 @@ def recorded_or( def require_may_change(self, todo: Todo) -> None: todo.require_commander(self) + def assigned(self, previous: Todo, assigned: Todo) -> Todo: + return assigned + def selects_lookup(self, lookup: str) -> bool: return False @@ -228,6 +231,9 @@ class StallWatcherIncarnation(BirthlessIncarnation): def require_may_change(self, todo: Todo) -> None: """The Core watcher may advance stalled orders without a commander turn.""" + def assigned(self, previous: Todo, assigned: Todo) -> Todo: + return previous.reassigned(assigned, self) + @classmethod def watcher(cls) -> StallWatcherIncarnation: """Named by its class: ``StallWatcherIncarnation`` is ``stall_watcher``.""" diff --git a/src/agent_comms/todos.py b/src/agent_comms/todos.py index 1e6b47dc45..e425d24b3a 100644 --- a/src/agent_comms/todos.py +++ b/src/agent_comms/todos.py @@ -37,7 +37,7 @@ from .channel_targets import CommandTarget from .goals import CountCheckResult, OrderCheck, ProblemGoalOrder from .recorded_time import EventTime, RecordedEventTime, UnrecordedEventTime -from .thread_identity import RecordedThreadIncarnation, StallWatcherIncarnation, ThreadIncarnation +from .thread_identity import RecordedThreadIncarnation, ThreadIncarnation from .threads import Thread from .typed_table import Column, Index, JsonStorage, TypedStore, TypedTable from .standing_rules import UndatedStandingRule @@ -497,10 +497,6 @@ def distance_readings(self) -> DistanceReadings: """The order's distance readings; progress that read no commit has none.""" return DistanceReadings() - def advance(self, todo: Todo, ordered: float, now: float) -> Todo: - """Progress with no stall consequence leaves its order unchanged.""" - return todo - def stalled(self, ordered: float, now: float) -> bool: return False @@ -548,25 +544,12 @@ def state(self) -> str: ... @abstractmethod def last_move(self, ordered: float) -> float: ... - def stall_reason(self) -> str: - """The strike reason, with the window this class declares in ``seconds``.""" - return f"{self.state}; no commit, push or PR change for {self.seconds // 60} minutes" - def stall_at(self, ordered: float) -> float: return self.last_move(ordered) + self.seconds def stalled(self, ordered: float, now: float) -> bool: return now >= self.stall_at(ordered) - def advance(self, todo: Todo, ordered: float, now: float) -> Todo: - if not self.stalled(ordered, now): - return todo - owner, actor = todo.held(), StallWatcherIncarnation.watcher() - reason = StrikeReason(f"{owner.owner} stalled on {todo.id}: {self.stall_reason()}") - at = RecordedEventTime(now) - struck = StrikeTodo(todo.revision, actor, owner, reason.text, at).apply(todo) - return ReleaseTodo(struck.revision, actor, owner, reason, at).apply(struck) - def stalled_between(self, ordered: float, since: float, now: float) -> bool: return since < self.stall_at(ordered) <= now @@ -1769,8 +1752,9 @@ def apply(self, current: Todo) -> Todo: self.actor.require_may_change(current) assigned = (current if current.assignment == proposed else current.state.assign(current, proposed, self.traced)) - return replace(assigned, deadline=self.deadline, order=self.order, goal=self.goal, - revision=current.revision + 1) + assigned = replace(assigned, deadline=self.deadline, order=self.order, goal=self.goal, + revision=current.revision + 1) + return self.actor.assigned(current, assigned) @property def actor(self) -> ThreadIncarnation: @@ -2210,6 +2194,16 @@ def assigned(self, proposed: Assignment, traced: FileLine | None = None) -> Todo proposed = self.assignment.successor(proposed, self) return replace(self, assignment=proposed, revision=self.revision + 1) + def reassigned(self, assigned: Todo, actor: ThreadIncarnation) -> Todo: + """After assignment validation, fold its strike and release into the same row write.""" + owner, at = self.held(), RecordedEventTime(time.time()) + reason = StrikeReason(f"Core reassigned stalled {self.id} from {owner.owner} " + f"to {assigned.held().owner}.") + struck = StrikeTodo(self.revision, actor, owner, reason.text, at).apply(self) + released = ReleaseTodo(struck.revision, actor, owner, reason, at).apply(struck) + return replace(assigned, strikes=struck.strikes, released=released.released, + revision=released.revision + 1) + def require_released_order(self, owner: ThreadIncarnation, goal_id: str) -> None: if self.assignment is not None and self.assignment.owned_by(owner, goal_id): raise TodoConflict(self) @@ -2404,30 +2398,14 @@ 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 advance_stalled(self, storage: StorageRule) -> tuple[Todo, ...]: - """Advance active orders; retry only rows released by this watcher, never other backlog.""" + def stalled(self, storage: StorageRule) -> tuple[Todo, ...]: + """Observe active held orders without changing owners, strikes or releases.""" held = tuple(todo for todo in self._open( "assignment IS NOT NULL", (), OpenTodoState) if todo.pending) pulls = PullRequests(held) - for todo in held: - observed_at = time.time() - source = ProgressSource.of(todo, storage, pulls) - progress = OrderProgress.from_source(source) - try: - with self.writing() as db: - current = self._current(db, todo.id) - current.require_result(todo) - result = progress.advance(current, source.since, observed_at) - if result != current: - result.upsert(db) - except TodoConflict: - # A concurrent order change owns the row; observe its new assignment next pass. - continue - watcher = StallWatcherIncarnation.watcher() - return self._open( - "assignment IS NULL AND json_extract(released,'$[#-1].actor.name')=? " - "AND json_extract(released,'$[#-1].actor.created_at')=?", - (watcher.name, watcher.wire_birth), OpenTodoState) + return tuple(todo for todo in held + if OrderProgress.from_source(source := ProgressSource.of(todo, storage, pulls)) + .stalled(source.since, time.time())) def reconcile_merges(self, commander: ThreadIncarnation, repos: Sequence[Repo], storage: StorageRule) -> None: diff --git a/tests/test_stall_watcher_strike.py b/tests/test_stall_watcher_strike.py index 20e14451fd..4f66336bd5 100644 --- a/tests/test_stall_watcher_strike.py +++ b/tests/test_stall_watcher_strike.py @@ -1,14 +1,22 @@ import asyncio import json import shutil +import subprocess +import sys +from dataclasses import replace from contextlib import contextmanager from pathlib import Path +import pytest + from agent_comms.comms import Comms from agent_comms.commander import standing from agent_comms.commander.config import CommanderConfig +from agent_comms.connector import Connector +from agent_comms.connector_store import ConnectorStore from agent_comms.thread_identity import StallWatcherIncarnation -from agent_comms.todos import BlockedTodoState, ChangeTodoState, ReleaseTodo, StrikeReason, StrikeTodo +from agent_comms.todos import (BlockedTodoState, ChangeTodoState, CorrectCheckTodo, + ReleaseTodo, StrikeReason, StrikeTodo, TodoConflict) from test_commander_problem_orders import create from test_standing_rules import working @@ -86,13 +94,21 @@ def test_parked_units_come_from_the_existing_commander_configuration(commander, assert standing.WheelSpin.idle(comms, "lead") == [] -def test_blocked_held_orders_are_not_struck_or_released(commander): - comms, _ = commander +def test_blocked_held_orders_are_not_struck_or_released(commander, commander_script): + original, _ = commander + original.registry.declare(original.registry.new_thread( + 'spare', frozenset({'comms'}), str(original.root), created_at=3.0)) todo = create(commander, 'P1') - blocked = comms.goals.todos.apply('P1', ChangeTodoState( - todo.revision, BlockedTodoState, comms.registry.require('lead').incarnation, None)) + blocked = original.goals.todos.apply('P1', ChangeTodoState( + todo.revision, BlockedTodoState, original.registry.require('lead').incarnation, None)) + copied = original.root.with_name('blocked-copy') + shutil.copytree(original.root, copied) + comms = Comms(copied) standing.UnownedWheelSpin.after_stall(comms) - assert comms.goals.todos.get('P1') == blocked + with pytest.raises(TodoConflict): + standing.UnownedWheelSpin(blocked, 'spare').assign( + comms, 'lead', comms.registry.require('spare'), StallWatcherIncarnation.watcher()) + assert comms.goals.todos.get('P1') == blocked == original.goals.todos.get('P1') def test_commander_released_backlog_is_not_automatically_reassigned(commander): @@ -105,12 +121,11 @@ def test_commander_released_backlog_is_not_automatically_reassigned(commander): assert comms.goals.todos.get('P1') == released -def test_watcher_release_waits_for_a_spare_without_striking_twice(commander, commander_script): +def test_no_spare_leaves_the_old_holder_and_strikes_untouched(commander, commander_script): comms, _ = commander - create(commander, 'P1') + original = create(commander, 'P1') standing.UnownedWheelSpin.after_stall(comms) - waiting = comms.goals.todos.get('P1') - assert waiting.holder is None and len(waiting.strikes) == len(waiting.released) == 1 + assert comms.goals.todos.get('P1') == original comms.registry.declare(comms.registry.new_thread("spare", frozenset({'comms'}), str(comms.root), created_at=3.0)) working(comms, 'worker', 2.0) working(comms, 'spare', 3.0) @@ -118,3 +133,72 @@ def test_watcher_release_waits_for_a_spare_without_striking_twice(commander, com assigned = comms.goals.todos.get('P1') assert assigned.holder == 'spare' assert len(assigned.strikes) == len(assigned.released) == 1 + + +def test_refused_reassignment_keeps_the_original_owner_and_strikes(commander, commander_script, tmp_path): + comms, _ = commander + todo = create(commander, 'P1') + comms.registry.declare(comms.registry.new_thread('spare', frozenset({'comms'}), str(comms.root), created_at=3.0)) + working(comms, 'spare', 3.0) + checks = tmp_path / 'checks' + checks.mkdir() + before = comms.goals.todos.apply('P1', CorrectCheckTodo( + todo.revision, comms.registry.require('lead').incarnation, todo.order.check, + replace(todo.order.check, directory=str(checks)), 'Use this check directory for the original order.')) + copied = comms.root.with_name('refused-copy') + shutil.copytree(comms.root, copied) + checks.rmdir() + with pytest.raises(ValueError): + standing.UnownedWheelSpin.after_stall(Comms(copied)) + assert Comms(copied).goals.todos.get('P1') == before == comms.goals.todos.get('P1') + + +async def test_three_connector_passes_allow_real_store_maintenance_between_passes( + commander, commander_script, tmp_path, monkeypatch): + original, _ = commander + before = create(commander, 'P1') + original.registry.declare(original.registry.new_thread( + 'spare', frozenset({'comms'}), str(original.root), created_at=3.0)) + working(original, 'spare', 3.0) + copied = original.root.with_name('three-pass-copy') + shutil.copytree(original.root, copied) + connector = Connector(ConnectorStore.at(tmp_path / 'auth'), 'http://127.0.0.1:18766', copied) + done, receipts = asyncio.Event(), [] + maintenance = ''' +import fcntl, os, sys +from pathlib import Path +from agent_comms.active_route import active_route_path +from agent_comms.maintenance_barrier import MaintenanceBarrier, MaintenanceOperator +from agent_comms.store_files import _store_lock +barrier = MaintenanceBarrier(Path(sys.argv[1]) / 'registry.json') +held = barrier.begin(MaintenanceOperator('P717 copied-root proof')) +try: + with _store_lock(barrier.wire_path), _store_lock(barrier.registry_path): + directory = os.open(active_route_path().parent, os.O_RDONLY | os.O_DIRECTORY) + try: + fcntl.flock(directory, fcntl.LOCK_EX | fcntl.LOCK_NB) + print('maintenance acquired wire, registry and route locks') + finally: + os.close(directory) +finally: + barrier.clear(held) +''' + + async def between(seconds): + run = subprocess.run([sys.executable, '-B', '-c', maintenance, str(copied)], + capture_output=True, text=True, timeout=10) + assert run.returncode == 0, run.stderr + receipts.append(run.stdout.strip()) + if len(receipts) == 3: + done.set() + await asyncio.Future() + + monkeypatch.setattr(standing.asyncio, 'sleep', between) + async with connector.watch_orders(): + await asyncio.wait_for(done.wait(), timeout=30) + after = Comms(copied).goals.todos.get('P1') + assert after.holder == 'spare' and len(after.strikes) == len(after.released) == 1 + assert original.goals.todos.get('P1') == before + (tmp_path / 'three-pass-result.json').write_text(json.dumps({ + 'copy': str(copied), 'passes': 3, 'maintenance': receipts, + 'holder': after.holder, 'strikes': len(after.strikes), 'original_unchanged': True}, indent=2)) From dcf9fc805536dd183c972da836a3c3df6caa9727 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Mon, 19 Oct 2026 19:33:49 -0400 Subject: [PATCH 16/18] Complete assignment callers and correct the copied-root verification cases --- src/agent_comms/todos.py | 4 ++-- tests/test_commander_run_rules.py | 4 ++-- tests/test_stall_watcher_strike.py | 2 +- tests/test_todos.py | 2 +- 4 files changed, 6 insertions(+), 6 deletions(-) diff --git a/src/agent_comms/todos.py b/src/agent_comms/todos.py index e425d24b3a..7ac53b9284 100644 --- a/src/agent_comms/todos.py +++ b/src/agent_comms/todos.py @@ -2197,8 +2197,8 @@ def assigned(self, proposed: Assignment, traced: FileLine | None = None) -> Todo def reassigned(self, assigned: Todo, actor: ThreadIncarnation) -> Todo: """After assignment validation, fold its strike and release into the same row write.""" owner, at = self.held(), RecordedEventTime(time.time()) - reason = StrikeReason(f"Core reassigned stalled {self.id} from {owner.owner} " - f"to {assigned.held().owner}.") + reason = StrikeReason(f"{owner.owner} stopped making progress on {self.id}; " + f"Core reassigned the order to {assigned.held().owner}.") struck = StrikeTodo(self.revision, actor, owner, reason.text, at).apply(self) released = ReleaseTodo(struck.revision, actor, owner, reason, at).apply(struck) return replace(assigned, strikes=struck.strikes, released=released.released, diff --git a/tests/test_commander_run_rules.py b/tests/test_commander_run_rules.py index ccfd7739fd..8155378f5c 100644 --- a/tests/test_commander_run_rules.py +++ b/tests/test_commander_run_rules.py @@ -31,11 +31,11 @@ def test_clean_transfers_keep_history_and_third_owner_needs_a_real_line(tmp_path creator=lead, owner=lead, generation='one', goal=GoalRef('lead', 1, 'one'), deadline=RecordedEventTime(1000), order=order(tmp_path)) second = store.assign('P1', expected_revision=first.revision, - owner=thread('second', tmp_path, 2), assigner=lead.incarnation, operation=AssignTodo, generation='two', goal=GoalRef('second', 2, 'two'), + owner=thread('second', tmp_path, 2), assigner=lead.incarnation, generation='two', goal=GoalRef('second', 2, 'two'), deadline=RecordedEventTime(1000), order=order(tmp_path)) assert not second.strikes kwargs = dict(expected_revision=second.revision, owner=thread('third', tmp_path, 3), - assigner=lead.incarnation, operation=AssignTodo, generation='three', goal=GoalRef('third', 3, 'three'), deadline=RecordedEventTime(1000), order=order(tmp_path)) + assigner=lead.incarnation, generation='three', goal=GoalRef('third', 3, 'three'), deadline=RecordedEventTime(1000), order=order(tmp_path)) with pytest.raises(ValueError, match='--traced PATH:LINE'): store.assign('P1', **kwargs) with pytest.raises(ValueError, match='does not exist'): diff --git a/tests/test_stall_watcher_strike.py b/tests/test_stall_watcher_strike.py index 4f66336bd5..99cdc6a8dc 100644 --- a/tests/test_stall_watcher_strike.py +++ b/tests/test_stall_watcher_strike.py @@ -194,7 +194,7 @@ async def between(seconds): await asyncio.Future() monkeypatch.setattr(standing.asyncio, 'sleep', between) - async with connector.watch_orders(): + async with connector.watch_orders(connector.application()): await asyncio.wait_for(done.wait(), timeout=30) after = Comms(copied).goals.todos.get('P1') assert after.holder == 'spare' and len(after.strikes) == len(after.released) == 1 diff --git a/tests/test_todos.py b/tests/test_todos.py index 2e9aaf1cad..13c9d61363 100644 --- a/tests/test_todos.py +++ b/tests/test_todos.py @@ -43,7 +43,7 @@ def _assign_process(root: str, name: str, created: float, gate) -> tuple[str, st try: row = store.assign( "todo-1", expected_revision=1, - owner=thread(name, Path(root), created), assigner=thread("lead", Path(root), 1.0).incarnation, operation=AssignTodo, + owner=thread(name, Path(root), created), assigner=thread("lead", Path(root), 1.0).incarnation, generation=f"generation-{name}", goal=GoalRef("lead", 1.0, "goal-1"), deadline=RecordedEventTime(1000.0), order=order(Path(root)), ) From 7869bb38e759916abe74a8805bd70b963ceddee8 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Mon, 19 Oct 2026 19:36:28 -0400 Subject: [PATCH 17/18] Register the real connector caller in the copied maintenance proof --- tests/test_stall_watcher_strike.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/tests/test_stall_watcher_strike.py b/tests/test_stall_watcher_strike.py index 99cdc6a8dc..b1d4cf82cd 100644 --- a/tests/test_stall_watcher_strike.py +++ b/tests/test_stall_watcher_strike.py @@ -160,6 +160,8 @@ async def test_three_connector_passes_allow_real_store_maintenance_between_passe original.registry.declare(original.registry.new_thread( 'spare', frozenset({'comms'}), str(original.root), created_at=3.0)) working(original, 'spare', 3.0) + original.registry.declare(original.registry.new_thread( + 'claude-ai', frozenset(), str(original.root), created_at=4.0)) copied = original.root.with_name('three-pass-copy') shutil.copytree(original.root, copied) connector = Connector(ConnectorStore.at(tmp_path / 'auth'), 'http://127.0.0.1:18766', copied) From 2b2fefd4cd4d4838de75c5da3217f6936df7d722 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Mon, 19 Oct 2026 19:38:30 -0400 Subject: [PATCH 18/18] Exercise three watcher passes through the real connector application lifecycle --- tests/test_stall_watcher_strike.py | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/tests/test_stall_watcher_strike.py b/tests/test_stall_watcher_strike.py index b1d4cf82cd..5de9589f66 100644 --- a/tests/test_stall_watcher_strike.py +++ b/tests/test_stall_watcher_strike.py @@ -8,8 +8,10 @@ from pathlib import Path import pytest +from aiohttp.test_utils import TestClient, TestServer from agent_comms.comms import Comms +from agent_comms.active_route import active_route_path from agent_comms.commander import standing from agent_comms.commander.config import CommanderConfig from agent_comms.connector import Connector @@ -154,7 +156,7 @@ def test_refused_reassignment_keeps_the_original_owner_and_strikes(commander, co async def test_three_connector_passes_allow_real_store_maintenance_between_passes( - commander, commander_script, tmp_path, monkeypatch): + commander, commander_script, tmp_path, monkeypatch, caplog): original, _ = commander before = create(commander, 'P1') original.registry.declare(original.registry.new_thread( @@ -195,9 +197,11 @@ async def between(seconds): done.set() await asyncio.Future() + active_route_path().parent.mkdir(parents=True, exist_ok=True) monkeypatch.setattr(standing.asyncio, 'sleep', between) - async with connector.watch_orders(connector.application()): + async with TestClient(TestServer(connector.application())): await asyncio.wait_for(done.wait(), timeout=30) + assert 'Stall watcher pass failed' not in caplog.text after = Comms(copied).goals.todos.get('P1') assert after.holder == 'spare' and len(after.strikes) == len(after.released) == 1 assert original.goals.todos.get('P1') == before