diff --git a/src/agent_comms/commander/order.py b/src/agent_comms/commander/order.py index 3a04251f09..dcd77ac396 100644 --- a/src/agent_comms/commander/order.py +++ b/src/agent_comms/commander/order.py @@ -297,7 +297,8 @@ def send(unit, due, body: str, clean=False, mechanical=False, reassign=None, new problem=new[0], repo=Repo.of(Path(check.directory)).github, text=new[1], **common))) for pid in [] if new else group: call(CommsOrderTool(owner=unit, order=OrderGoalAction( - problem=pid, expected_revision=COMMS.goals.todos.get(pid).revision, traced=trace, **common))) + problem=pid, expected_revision=COMMS.goals.todos.get(pid).revision, + assigner=COMMS.registry.require(ME).incarnation, traced=trace, **common))) message += "\n" + "\n".join(order.pull_request_instruction(pid) for pid in group) sent = post(VerdictAudience((unit,)), message) # The message is each order's recorded link; a verdict citing it lands on that order. @@ -395,8 +396,8 @@ def board(): "GENERATED by agent-comms-commander order from Core's Todo store. Do not edit by hand:", '`order problem` adds a row, `order send ... Problem: P` orders it, `order close` or `order withdraw` removes it.', '', '| Problem | Owner file | Owner | Order | Due | Progress | Delivered ' - '| Owner last acted |', - '|---|---|---|---|---|---|---|---|'] + '| Strikes | Owner last acted |', + '|---|---|---|---|---|---|---|---|---|'] listed, read = COMMANDER.orders(CONFIG.repos), progress() activity = {unit: last_activity(unit) for unit in {t.holder for t in listed if t.holder}} for t in sorted(listed, key=lambda t: int(t.id[1:]) if t.id[1:].isdigit() else 0): @@ -405,7 +406,7 @@ def board(): lines.append(f'| {t.id}: {t.text} | {t.order.owner.to_wire()} | {owner} | {t.order.id} | ' f'{hhmm(t.deadline)} | {read[t.id].line if t.id in read else "-"} | ' f'{"no" if t.pending else "yes, install pending"} | ' - f'{activity[unit].clock() if unit else "-"} |') + f'{t.strike_count} | {activity[unit].clock() if unit else "-"} |') history = COMMS.goals.todos.history_for_commander(COMMS.registry.require(ME).incarnation) lines += ['', '## Withdrawn orders', '', *(f'- {line}' for t in history for line in t.order.withdrawals(t))] @@ -590,20 +591,18 @@ def replied(latest: Message | None) -> str: def deadline_lines(pending: list[Todo], read: dict[str, OrderProgress], since: float = 0.0) -> list[str]: """Past-due unfinished orders alarm; installed merges have already closed. - Stall crossings remain separate from deadline alarms.""" + Core records a stalled order's strike and release; this view reports deadlines.""" now = time.time() - late, stalled = [], [] + late = [] for t in pending: ordered, work = t.order.at.recorded_at('An order'), read[t.id] - if work.stalled_between(ordered, since, now): - stalled.append(f'STALL {t.id} {t.holder}: {work.line} -> push it now or reassign') if not t.overdue(now): continue answer = reply(t.holder, ordered) due = f'due {hhmm(t.deadline)}, order {t.order.id}' late.append(f'{t.holder} ({due}, {work.line}{replied(answer)})') alarm = f'OVERDUE: {", ".join(late)} -> push or merge unfinished work now' - return [alarm] * bool(late) + stalled + return [alarm] * bool(late) def install_lines(held: tuple[Todo, ...], read: dict[str, OrderProgress], now: float) -> list[str]: @@ -634,12 +633,6 @@ def overdue(): for line in lines: print(line, flush=True) installs = lines - # Units leave finished PRs open and wander off. An order's open PR that nobody has touched for - # 5 minutes is a stall, whatever the unit says it is doing; until its deadline, OVERDUE owns the alarm. - idle_prs = [f'{t.holder} {t.repo.split("/")[1]} {read[t.id].line} ({t.id})' - for t in pending if t.overdue(now) and read[t.id].untouched(now)] - if idle_prs: - print('STALL PR open and untouched 5+ min: ' + ', '.join(idle_prs) + ' -> order the merge or the fix now', flush=True) for _ in HuntDue.broken(COMMS, ME, since): print(HuntVerb.due(next_hunt()), flush=True) since = now diff --git a/src/agent_comms/commander/standing.py b/src/agent_comms/commander/standing.py index ddc65d71e8..6f56bb955d 100644 --- a/src/agent_comms/commander/standing.py +++ b/src/agent_comms/commander/standing.py @@ -2,14 +2,18 @@ from __future__ import annotations +import asyncio +import logging import subprocess import time -from collections.abc import Iterator +from collections.abc import Callable, Iterable, Iterator +from contextlib import AbstractContextManager from dataclasses import dataclass from pathlib import Path from typing import ClassVar from ..comms import Comms +from ..coordinator import Coordination from ..goal_actions import OrderGoalAction from ..goals import Goal from ..standing_rules import ( @@ -18,14 +22,15 @@ StandingRule, UndatedStandingRule, ) +from ..thread_identity import StallWatcherIncarnation, ThreadIncarnation from ..threads import Thread -from ..todos import FlatTrend, OrderProgress, PullRequests, StrikeTodo, Todo +from ..todos import ( + FlatTrend, OrderProgress, PullRequests, Todo, +) +from ..tools import CommsUnjudgedTool from ..turn_digests import TurnDigestRead, TurnsSince -from . import judge from .config import CommanderConfig, Repo -CONFIG = CommanderConfig.installed() - @dataclass(frozen=True) class Detached(UndatedStandingRule): @@ -56,7 +61,8 @@ class RequestStalled(UndatedStandingRule): def violations(cls, comms: Comms, commander: str) -> tuple[RequestStalled, ...]: thread = comms.registry.require(commander) return tuple(cls(request) for goal in thread.goal.active() - for request, served in goal.order.open_requests(goal, comms.goals, thread.incarnation) + for request, served in goal.order.open_requests( + goal, comms.goals, thread.incarnation) if not served) def line(self) -> str: @@ -77,21 +83,22 @@ class WheelSpin(StandingRule, declared_name=None): def idle(cls, comms: Comms, commander: str) -> list[Thread]: """Units with a recent turn in their own transcript and no order, parked threads aside.""" held = {todo.holder for todo in cls.todos(comms, commander)} - excluded = held | CONFIG.parked | {commander, ""} + excluded = held | CommanderConfig.installed().parked | {commander, ""} bound = TurnsSince(time.time() - cls.recent) return sorted((thread for thread in comms.registry.all_threads().values() if thread.name not in excluded and TurnDigestRead(comms.root, thread, bound).read().turns), key=lambda thread: thread.name) - def assign(self, comms: Comms, commander: str, unit: Thread) -> None: + def assign(self, comms: Comms, commander: str, unit: Thread, + assigner: ThreadIncarnation) -> None: """The order sent again to ``unit`` through its Goal, with the order's own text.""" - todo = comms.goals.todos.get(self.todo.id) + todo = self.todo order = todo.order.reissued(time.time()) comms.goals.update_goal(unit.name, OrderGoalAction( problem=todo.id, expected_revision=todo.revision, order=order, request=todo.goal.request(), deadline=todo.deadline, commander=commander, - traced=todo.order.owner)) + assigner=assigner, traced=todo.order.owner)) comms.messaging.send(commander, unit.name, f"{self.line()}\n{order.instructions(todo)}") @@ -122,15 +129,8 @@ def line(self) -> str: " to an idle unit") def enforce(self, comms: Comms, commander: str) -> None: - todo = comms.goals.todos.get(self.todo.id) - held = todo.held() - if not todo.struck_on_order(held.owner): - comms.goals.change_problem(todo.id, StrikeTodo( - todo.revision, comms.registry.require(commander).incarnation, held, - f"{held.owner} kept working on {todo.id} while the distance to done did not fall" - " across two readings"), commander) for unit in self.idle(comms, commander)[:1]: - self.assign(comms, commander, unit) + self.assign(comms, commander, unit, StallWatcherIncarnation.watcher()) @dataclass(frozen=True) @@ -141,17 +141,57 @@ class UnownedWheelSpin(WheelSpin, UndatedStandingRule): @classmethod def violations(cls, comms: Comms, commander: str) -> tuple[UnownedWheelSpin, ...]: - rows = (todo for todo in comms.goals.todos.for_commander( - comms.registry.require(commander).incarnation) if todo.pending and not todo.holder) - idle = cls.idle(comms, commander) - return tuple(cls(todo, unit.name) for unit, todo in zip(idle, rows, strict=False)) + return cls.paired(comms, commander, (todo for todo in + comms.goals.todos.for_commander(comms.registry.require(commander).incarnation) + if not todo.holder)) + + @classmethod + def paired(cls, comms: Comms, commander: str, + todos: Iterable[Todo]) -> tuple[UnownedWheelSpin, ...]: + """Each pending row with the first idle unit not struck on it.""" + idle, found = cls.idle(comms, commander), [] + for todo in (todo for todo in todos if todo.pending): + for unit in idle: + if not todo.live_strikes(unit.name): + idle.remove(unit) + found.append(cls(todo, unit.name)) + break + return tuple(found) def line(self) -> str: return (f"WHEEL SPIN: {self.unit} holds no order while {self.todo.id} is unowned" f" -> Core gives it to {self.unit}") def enforce(self, comms: Comms, commander: str) -> None: - self.assign(comms, commander, comms.registry.require(self.unit)) + self.assign(comms, commander, comms.registry.require(self.unit), + comms.registry.require(commander).incarnation) + + @classmethod + def after_stall(cls, comms: Comms) -> None: + """Assign first: refused or unavailable replacements leave the old holder untouched.""" + stalled = comms.goals.todos.stalled(comms.registry.storage_rule) + for commander in {todo.creator_identity.name for todo in stalled}: + mine = (todo for todo in stalled if todo.creator_identity.name == commander) + for rule in cls.paired(comms, commander, mine): + rule.assign(comms, commander, comms.registry.require(rule.unit), + StallWatcherIncarnation.watcher()) + + @classmethod + def watch_pass(cls, opened_wire: Callable[[], AbstractContextManager[Comms]]) -> None: + """The worker borrows the root for this pass, never while the timer is idle.""" + with opened_wire() as comms: + cls.after_stall(comms) + + @classmethod + async def watch(cls, opened_wire: Callable[[], AbstractContextManager[Comms]]) -> None: + """Run a pass each minute; failures go to the existing service log.""" + while True: + (outcome,) = await asyncio.gather( + Coordination.run_worker(lambda: cls.watch_pass(opened_wire)), + return_exceptions=True) + if isinstance(outcome, Exception): + logging.error("Stall watcher pass failed", exc_info=outcome) + await asyncio.sleep(60) @dataclass(frozen=True) @@ -161,7 +201,7 @@ class Unjudged(ReasonedStandingRule, UndatedStandingRule): @classmethod def reason(cls, comms: Comms, commander: str) -> str | None: - total = len(judge.Edit.unjudged(comms, commander)) + total = CommsUnjudgedTool.count(comms, commander) return str(total) if total > cls.limit else None @@ -224,7 +264,8 @@ def merges(cls, repo: Repo) -> Iterator[MergeGrowth]: @classmethod def violations(cls, comms: Comms, commander: str) -> tuple[MergeGrowth, ...]: - return tuple(growth for repo in CONFIG.repos for growth in cls.merges(repo) if growth.over) + return tuple(growth for repo in CommanderConfig.installed().repos + for growth in cls.merges(repo) if growth.over) @property def over(self) -> bool: diff --git a/src/agent_comms/connector.py b/src/agent_comms/connector.py index 9ff7e81d2a..7b37fdecc9 100644 --- a/src/agent_comms/connector.py +++ b/src/agent_comms/connector.py @@ -27,8 +27,8 @@ import time import urllib.request from abc import abstractmethod -from collections.abc import Iterator, Mapping, Sequence -from contextlib import contextmanager +from collections.abc import AsyncIterator, Iterator, Mapping, Sequence +from contextlib import contextmanager, suppress from dataclasses import dataclass, field, fields, replace from pathlib import Path from typing import Annotated, ClassVar, Literal, Self, get_args @@ -49,6 +49,7 @@ RegisteredClient, digest, ) from .messages import Message +from .commander.standing import UnownedWheelSpin from .todos import DirectiveText from .tool_caller import ThreadCaller from .tools import CommsChannelsTool, ReadTool, ToolRequest @@ -359,6 +360,7 @@ def application(self) -> web.Application: with self.opened_wire(): pass app = web.Application() + app.cleanup_ctx.append(self.watch_orders) app.add_routes( [ web.get("/.well-known/oauth-protected-resource", self.resource_metadata), @@ -372,6 +374,16 @@ def application(self) -> web.Application: ) return app + async def watch_orders(self, app: web.Application) -> AsyncIterator[None]: + """The running Core service advances stalled orders without a commander call.""" + task = asyncio.create_task(UnownedWheelSpin.watch(self.opened_wire)) + try: + yield + finally: + task.cancel() + with suppress(asyncio.CancelledError): + await task + async def resource_metadata(self, request: web.Request) -> web.Response: return web.json_response( { 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/goal_actions.py b/src/agent_comms/goal_actions.py index e5ca5b006f..c011d6f757 100644 --- a/src/agent_comms/goal_actions.py +++ b/src/agent_comms/goal_actions.py @@ -27,17 +27,18 @@ from .goal_waits import GoalWaits, RecordedGoalWait from .goals import ( Goal, - ReportedTurn, GoalDeclaration, GoalOrder, NoGoal, ProblemGoalOrder, + ReportedTurn, RequestDirective, RequestGoalOrder, UnorderedGoalOrder, ) from .progress_claims import ProgressClaim from .recorded_time import EventTime +from .thread_identity import ThreadIncarnation from .threads import Thread from .todos import GoalRef, PendingProblemOrder, SendBackVerdict, Todo, TodoOperation @@ -507,6 +508,9 @@ class ProblemOrderGoalAction(GoalAction[Goal], RuntimeInvocable): deadline: EventTime commander: str + @abstractmethod + def bound_to(self, actor: ThreadIncarnation) -> ProblemOrderGoalAction: ... + @abstractmethod def reserve(self, ctx: GoalActionContext, generation: str, goal: GoalRef) -> Todo: ... @@ -542,12 +546,17 @@ def create(self, ctx: GoalActionContext) -> Goal: @dataclass(frozen=True, kw_only=True) class OrderGoalAction(ProblemOrderGoalAction): + """The order's commander gives the problem to this thread.""" expected_revision: int traced: FileLine | None = field(default=None, metadata={"wire_omit_default": True}) + assigner: ThreadIncarnation + + def bound_to(self, actor: ThreadIncarnation) -> OrderGoalAction: + return replace(self, commander=actor.name, assigner=actor) def reserve(self, ctx: GoalActionContext, generation: str, goal: GoalRef) -> Todo: return ctx.goals.todos.assign(self.problem, expected_revision=self.expected_revision, - owner=ctx.thread, parent=ctx.goals.registry.require(self.commander), + owner=ctx.thread, assigner=self.assigner, generation=generation, goal=goal, deadline=self.deadline, order=self.order, traced=self.traced) @@ -557,6 +566,9 @@ class CreateProblemGoalAction(ProblemOrderGoalAction): repo: str text: str + def bound_to(self, actor: ThreadIncarnation) -> CreateProblemGoalAction: + return replace(self, commander=actor.name) + def reserve(self, ctx: GoalActionContext, generation: str, goal: GoalRef) -> Todo: return ctx.goals.todos.create(self.problem, self.repo, self.text, creator=ctx.goals.registry.require(self.commander), owner=ctx.thread, diff --git a/src/agent_comms/thread_identity.py b/src/agent_comms/thread_identity.py index 366cda2f82..6b20289488 100644 --- a/src/agent_comms/thread_identity.py +++ b/src/agent_comms/thread_identity.py @@ -17,7 +17,7 @@ from .agent_backend import AgentBackend, RuntimeLocation from .channel_targets import Tag 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 @@ -30,6 +30,7 @@ from .read_ledger import ReadDocument, ReadLedger from .registry_provenance import RegistryProvenance from .threads import Thread + from .todos import Todo T = TypeVar("T") @@ -91,6 +92,12 @@ def recorded_or( ) -> T: return otherwise() + 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 @@ -162,52 +169,79 @@ 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: object) -> 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 + + 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``.""" + return cls(declared_words(cls.__name__.removesuffix("Incarnation"))) @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,10 +307,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) + if self.created_at in BirthlessIncarnation.members: + return BirthlessIncarnation.members[self.created_at](self.name) return RecordedThreadIncarnation(self.name, self.created_at) @classmethod diff --git a/src/agent_comms/todos.py b/src/agent_comms/todos.py index 19c45babcb..7ac53b9284 100644 --- a/src/agent_comms/todos.py +++ b/src/agent_comms/todos.py @@ -18,7 +18,7 @@ from abc import abstractmethod from collections.abc import Iterator, Sequence from dataclasses import dataclass, field, fields, replace -from functools import cache +from functools import cache, cached_property from importlib.resources import files from pathlib import Path from typing import TYPE_CHECKING, ClassVar, Final, Self @@ -355,6 +355,9 @@ def commits(self, since: float, *excluded: str) -> Iterator[tuple[str, float]]: @abstractmethod def readings(self, source: ProgressSource) -> DistanceReadings: ... + @abstractmethod + def pushes(self, since: float) -> Iterator[tuple[str, float]]: ... + class MissingCheckout(ProgressCheckout): """A retired checkout contributes no local history, not a failed Git read.""" @@ -365,6 +368,9 @@ def commits(self, since: float, *excluded: str) -> Iterator[tuple[str, float]]: def readings(self, source: ProgressSource) -> DistanceReadings: return DistanceReadings() + def pushes(self, since: float) -> Iterator[tuple[str, float]]: + return iter(()) + class PresentCheckout(ProgressCheckout): def git(self, *args: str) -> str: @@ -393,6 +399,17 @@ def readings(self, source: ProgressSource) -> DistanceReadings: return DistanceReadings( tuple(source.reading(self, base, sha, at) for sha, at in points)) + def pushes(self, since: float) -> Iterator[tuple[str, float]]: + """A push-ref update has its own time, not the commit's author time.""" + branch = self.git("branch", "--show-current").strip() + after = time.strftime('%Y-%m-%dT%H:%M:%SZ', time.gmtime(since)) + for row in self.git("reflog", "show", "--all", "--date=unix", f"--since={after}", + "--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}": + yield sha, float(stamp.removesuffix("}")) + @dataclass(frozen=True) class ProgressSource: @@ -419,8 +436,18 @@ def requests(self, state: str) -> list[PullRequest]: def commits(self, *excluded: str) -> Iterator[tuple[str, float]]: return self.checkout.commits(self.since, *excluded) + @cached_property + def pushes(self) -> tuple[PushedProgress, ...]: + return tuple(PushedProgress(sha[:9], at=at) for sha, at in self.checkout.pushes(self.since)) + + @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 readings(self) -> DistanceReadings: - return self.checkout.readings(self) + """Progress observation does not run the order's Check.""" + return DistanceReadings() def reading(self, checkout: PresentCheckout, base: str, sha: str, at: float) -> DistanceReading: """Keep the check command with its measured or missing distance.""" @@ -431,6 +458,13 @@ def reading(self, checkout: PresentCheckout, base: str, sha: str, at: float) -> self.todo.order.untouched(touched)), check.command) +class MeasuredProgressSource(ProgressSource): + """The board's distance view runs the Check; the stall watcher only reads activity.""" + + def readings(self) -> DistanceReadings: + return self.checkout.readings(self) + + @dataclass(frozen=True) class OrderProgress(DeclaredFamily, affix="Progress"): """Where an order's work stands, derived on each read from its unit's checkout and PRs. @@ -440,11 +474,14 @@ class OrderProgress(DeclaredFamily, affix="Progress"): @classmethod def read(cls, todo: Todo, storage: StorageRule, pulls: PullRequests) -> OrderProgress: """The first found member by its declared precedence; with none found, no commit yet.""" - source = ProgressSource.of(todo, storage, pulls) + return cls.from_source(MeasuredProgressSource.of(todo, storage, 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)), - NoCommitProgress()) + return next((progress for member in members + for progress in member.observations(source)), NoCommitProgress()) except subprocess.TimeoutExpired as unread: return UnreadProgress(str(unread)) @@ -490,28 +527,41 @@ 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: - """Work that stalls at a time its own class declares.""" + """One inactivity deadline and its strike/release transition.""" + + seconds: ClassVar[int] = 600 + @property @abstractmethod - def stall_at(self, ordered: float) -> float: ... + def state(self) -> str: ... + + @abstractmethod + def last_move(self, ordered: float) -> float: ... + + 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 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 distance_readings(self) -> DistanceReadings: return self.readings @@ -526,9 +576,8 @@ 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 @dataclass(frozen=True) @@ -540,9 +589,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 @@ -558,12 +616,14 @@ 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 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: @@ -571,15 +631,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: @@ -1124,6 +1184,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() @@ -1224,10 +1288,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: @@ -1548,9 +1608,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 @@ -1559,7 +1618,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) @@ -1675,6 +1735,7 @@ def apply(self, current: Todo) -> Todo: ... @dataclass(frozen=True) class AssignTodo(TodoOperation): + """The order's commander gives the problem to a unit.""" proposed: Assignment deadline: EventTime order: PendingProblemOrder @@ -1688,11 +1749,16 @@ 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.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: + return self.proposed.parent_identity @dataclass(frozen=True) @@ -1704,7 +1770,7 @@ class StrikeTodo(TodoOperation): def apply(self, current: Todo) -> Todo: current.require_revision(self.expected_revision) - current.require_commander(self.actor) + self.actor.require_may_change(current) current.require_assignment(self.previous) _text(self.reason, "Strike reason", 2000) StrikeReason(self.reason) @@ -1877,7 +1943,7 @@ class ReleaseTodo(TodoOperation): def apply(self, current: Todo) -> Todo: current.require_revision(self.expected_revision) - current.require_commander(self.actor) + self.actor.require_may_change(current) current.require_assignment(self.previous) return replace(current, assignment=None, released=(*current.released, self), revision=current.revision + 1) @@ -2078,7 +2144,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.""" @@ -2128,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"{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, + 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) @@ -2322,6 +2398,15 @@ 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 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) + 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: """Read before locking; close only orders whose observed revision still holds.""" @@ -2389,8 +2474,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)) @@ -2478,19 +2564,16 @@ def assign( *, expected_revision: int, owner: Thread, - parent: Thread, + assigner: ThreadIncarnation, generation: str, goal: GoalRef, deadline: EventTime, order: PendingProblemOrder, traced: FileLine | None = None, ) -> Todo: - identity, parent_identity = _thread(owner), _thread(parent) + identity = _thread(owner) 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.name, identity.created_at, assigner.name, assigner.wire_birth, generation) + return self.apply( + todo_id, AssignTodo(expected_revision, proposed, deadline, order, goal, traced)) + diff --git a/src/agent_comms/tools.py b/src/agent_comms/tools.py index 9400763226..b62ec09ce6 100644 --- a/src/agent_comms/tools.py +++ b/src/agent_comms/tools.py @@ -853,7 +853,8 @@ class CommsOrderTool(ThreadToolRequest): owner: str = tool_field("Assignee thread") def apply(self, comms: Comms, caller: ThreadCaller) -> ProblemOrdered: - goal = comms.goals.update_goal(self.owner, replace(self.order, commander=caller.name)) + goal = comms.goals.update_goal(self.owner, self.order.bound_to( + comms.registry.require(caller.name).incarnation)) return ProblemOrdered(goal, comms.goals.todos.get(self.order.problem)) diff --git a/tests/test_commander_problem_orders.py b/tests/test_commander_problem_orders.py index 6052ac9a2a..8bafefff6a 100644 --- a/tests/test_commander_problem_orders.py +++ b/tests/test_commander_problem_orders.py @@ -124,7 +124,7 @@ def test_existing_goal_reads_changed_todo_without_a_text_mirror(commander): replacement = order(comms.root, 'replacement', 'printf 7', 7) CommsOrderTool(owner='worker', order=OrderGoalAction(problem='P1', order=replacement, request='request', deadline=RecordedEventTime(3000), commander='lead', - expected_revision=todo.revision)).apply(comms, caller) + expected_revision=todo.revision, assigner=comms.registry.require('lead').incarnation)).apply(comms, caller) assert comms.registry.require('worker').goal.text == ProblemGoalOrder.text worker = comms.registry.require('worker').incarnation rows, rendered = comms.goals.todos.for_goal(worker, goal.id), '\n'.join(goal.instructions_for(comms.goals, worker)) @@ -1122,7 +1122,8 @@ def test_goal_stays_open_while_problems_pend(commander): other = comms.goals.todos.get('P3') CommsOrderTool(owner='worker', order=OrderGoalAction(problem='P2', order=order(comms.root, 'revised'), request='request', deadline=RecordedEventTime(3000), commander='lead', - expected_revision=comms.goals.todos.get('P2').revision)).apply(comms, caller) + expected_revision=comms.goals.todos.get('P2').revision, + assigner=comms.registry.require('lead').incarnation)).apply(comms, caller) assert comms.goals.todos.get('P2').order.id == 'revised' assert comms.goals.todos.get('P3') == other CommsOrderDoneTool(problem='P2', order_id=comms.goals.todos.get('P2').order.id, delivery=OperationDelivery(), claims=working(comms)).apply(comms, caller) @@ -1136,7 +1137,8 @@ def test_a_refused_send_leaves_the_todo_row_byte_identical(commander): with pytest.raises(ValueError, match='Finish or release the assigned problems'): CommsOrderTool(owner='worker', order=OrderGoalAction(problem='P3', order=order(comms.root, 'resent'), request='request', deadline=RecordedEventTime(3000), commander='lead', - expected_revision=comms.goals.todos.get('P3').revision)).apply(comms, caller) + expected_revision=comms.goals.todos.get('P3').revision, + assigner=comms.registry.require('lead').incarnation)).apply(comms, caller) assert todo_row(comms, 'P3') == before assert comms.registry.require('worker').goal == goal diff --git a/tests/test_commander_run_rules.py b/tests/test_commander_run_rules.py index 12475225d0..8155378f5c 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, 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, 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 new file mode 100644 index 0000000000..5de9589f66 --- /dev/null +++ b/tests/test_stall_watcher_strike.py @@ -0,0 +1,210 @@ +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 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 +from agent_comms.connector_store import ConnectorStore +from agent_comms.thread_identity import StallWatcherIncarnation +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 + + +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)) + 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) + standing.UnownedWheelSpin.after_stall(comms) + todo = comms.goals.todos.get("P1") + 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 todo.holder == "spare" and todo.live_strikes("worker") + 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 + (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 and controlled saved native records; actual history reader, no commander invocation."}, indent=2)) + + +def test_a_failed_stall_pass_logs_and_releases_the_root_before_the_next_pass(commander, monkeypatch, caplog): + comms, _ = commander + passes, borrowed = [], [] + + @contextmanager + def opened_wire(): + borrowed.append(comms) + try: + yield comms + finally: + borrowed.remove(comms) + + def after_stall(cls, c): + assert borrowed == [c] + passes.append(c) + if len(passes) == 1: + raise FileNotFoundError("no commander config") + + 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(opened_wire)) + except asyncio.CancelledError: + pass + 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, 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 = 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) + 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): + 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_no_spare_leaves_the_old_holder_and_strikes_untouched(commander, commander_script): + comms, _ = commander + original = create(commander, 'P1') + standing.UnownedWheelSpin.after_stall(comms) + 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) + standing.UnownedWheelSpin.after_stall(comms) + 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, caplog): + 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) + 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) + 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() + + active_route_path().parent.mkdir(parents=True, exist_ok=True) + monkeypatch.setattr(standing.asyncio, 'sleep', between) + 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 + (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)) 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 " diff --git a/tests/test_todos.py b/tests/test_todos.py index 1fe58fe3b9..13c9d61363 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, generation=f"generation-{name}", goal=GoalRef("lead", 1.0, "goal-1"), deadline=RecordedEventTime(1000.0), order=order(Path(root)), ) diff --git a/tests/test_tool_replies.py b/tests/test_tool_replies.py index ad26844ba8..e7a4e5206f 100644 --- a/tests/test_tool_replies.py +++ b/tests/test_tool_replies.py @@ -90,7 +90,8 @@ def scenario(comms: Comms, project, monkeypatch): yield CommsRuleReviewTool, {"entry": FieldCodec.encode(DrawnRuleReviewEntry(("Declare each fact once",), 0))}, lead yield CommsOrderTool, {"owner": "worker", "order": FieldCodec.encode(OrderGoalAction( problem="P1", order=order, request="request", deadline=RecordedEventTime(3000), - commander="lead", expected_revision=comms.goals.todos.get("P1").revision))}, lead + commander="lead", assigner=actor, + expected_revision=comms.goals.todos.get("P1").revision))}, lead yield CommsOrderCheckTool, {"problem": "P1", "check": FieldCodec.encode( OrderCheck(str(comms.root), "printf 1", CountCheckResult(1))), "why": "The recorded check counted the wrong file"}, lead