diff --git a/src/agent_comms/commander_heartbeat.py b/src/agent_comms/commander_heartbeat.py index 3d2f4f3cad..620c0c9e6e 100644 --- a/src/agent_comms/commander_heartbeat.py +++ b/src/agent_comms/commander_heartbeat.py @@ -5,10 +5,9 @@ from dataclasses import dataclass, field from typing import TYPE_CHECKING +from agent_comms.commander_service import CommanderService from agent_comms.heartbeat_frame import HeartbeatFrame from agent_comms.message_bus import WatchPass -from agent_comms.registration import Registration -from agent_comms.service_sender import ServiceSender from agent_comms.standing_rules import StandingRule from agent_comms.thread_identity import ThreadIncarnation from agent_comms.threads import Thread @@ -17,8 +16,7 @@ if TYPE_CHECKING: from agent_comms.comms import Comms - from agent_comms.goal_management import Goals - from agent_comms.todos import OwnerRequest, Todo + from agent_comms.todos import Todo @dataclass @@ -81,25 +79,9 @@ def read(cls, comms: Comms, thread: Thread, @dataclass -class CommanderHeartbeat(ServiceSender, WatchPass[tuple[CommanderObservation, ...]]): - sender = "commander-heartbeat" +class CommanderHeartbeat(CommanderService, WatchPass[tuple[CommanderObservation, ...]]): reviews: dict[ThreadIncarnation, CommanderReview] = field(default_factory=dict) - @classmethod - def requested_by(cls, thread: Thread, goals: Goals) -> tuple[OwnerRequest, ...]: - """The oldest open request makes a live holder owed service, even before its first beat.""" - return (goals.todos.open_requests(thread.incarnation, limit=1, descending=False) - if thread.process_alive else ()) - - @classmethod - def commanders(cls, registry: Registration, goals: Goals) -> tuple[Thread, ...]: - """The live request holders whose current incarnation owns commander work.""" - return tuple( - thread - for thread in registry.all_threads().values() - if cls.requested_by(thread, goals) - ) - def observe(self, comms: Comms) -> tuple[CommanderObservation, ...]: return tuple( CommanderObservation.read(comms, thread) diff --git a/src/agent_comms/commander_service.py b/src/agent_comms/commander_service.py new file mode 100644 index 0000000000..c533b6a7e1 --- /dev/null +++ b/src/agent_comms/commander_service.py @@ -0,0 +1,32 @@ +"""The commander service's sender and live request holders.""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +from .service_sender import ServiceSender +from .threads import Thread + +if TYPE_CHECKING: + from .goal_management import Goals + from .registration import Registration + from .todos import OwnerRequest + + +class CommanderService(ServiceSender): + sender = "commander-heartbeat" + + @classmethod + def requested_by(cls, thread: Thread, goals: Goals) -> tuple[OwnerRequest, ...]: + """The oldest open request makes a live holder owed service, even before its first beat.""" + return (goals.todos.open_requests(thread.incarnation, limit=1, descending=False) + if thread.process_alive else ()) + + @classmethod + def commanders(cls, registry: Registration, goals: Goals) -> tuple[Thread, ...]: + """The live request holders whose current incarnation owns commander work.""" + return tuple( + thread + for thread in registry.all_threads().values() + if cls.requested_by(thread, goals) + ) diff --git a/src/agent_comms/owner_lifecycle.py b/src/agent_comms/owner_lifecycle.py index 5e4086b25b..48cc2cbbe5 100644 --- a/src/agent_comms/owner_lifecycle.py +++ b/src/agent_comms/owner_lifecycle.py @@ -26,7 +26,7 @@ SynchronousProcess, ) from .cli_result import CliResult -from .commander_heartbeat import CommanderHeartbeat +from .commander_service import CommanderService from .diagnostics import OwnerProcessOutput from .errors import RelationViolationError from .launch_root import LaunchRoot @@ -335,7 +335,7 @@ def _register_owner( if previous.process_identity.offline(): orders = tuple( todo - for commander in CommanderHeartbeat.commanders(self.registry, self.goals) + for commander in CommanderService.commanders(self.registry, self.goals) for todo in self.goals.todos.held_by(commander.incarnation, owned.incarnation) if todo.pending and todo.held().owner_identity == owned.incarnation ) @@ -348,8 +348,8 @@ def _register_owner( def _announce_restart(self, name: str, identities: str) -> None: """Publish once after the registry's process reservation commits.""" - CommanderHeartbeat.register(self.registry) - self.messaging.send(CommanderHeartbeat.sender, name, + CommanderService.register(self.registry) + self.messaging.send(CommanderService.sender, name, f"Restarted holding {identities}; continue them from your branch") def ensure_owner(