diff --git a/src/agent_comms/closed_pull_requests.py b/src/agent_comms/closed_pull_requests.py index 046bf42c60..630d93c62c 100644 --- a/src/agent_comms/closed_pull_requests.py +++ b/src/agent_comms/closed_pull_requests.py @@ -14,7 +14,7 @@ PullRequests, StoredClosedHead, StoredClosedRepository, StoredCommitPatch, StoredHeadDiff, StoredMainPatch, ) -from .work_branches import WorkBranchError, WorkBranches +from .work_branches import CommitBase, UnservableCommitBase, WorkBranchError, WorkBranches if TYPE_CHECKING: from .comms import Comms @@ -61,10 +61,12 @@ class ClosedHeadScan: published: str new_patches: frozenset[str] fingerprints: dict[str, StoredCommitPatch] + unservable: tuple[str, ...] diffs: dict[tuple[str, str], StoredHeadDiff] = field(default_factory=dict) def head(self, request: PullRequestHead, closed_at: float) -> StoredClosedHead: - base = self.reader.commit_base(self.published, request.head) + base: CommitBase = (UnservableCommitBase() if request.head in self.unservable + else self.reader.commit_base(self.published, request.head)) return StoredClosedHead(request, base, closed_at, base.missing(self, request.head)) def missing_after(self, base: str, head: str) -> tuple[str, ...]: @@ -103,9 +105,12 @@ def observe_repository(store: TodoStore, repo: Repo) -> ClosedRepositoryObservat (published,) = reader.remote_heads(repo.remote, f'refs/heads/{reader.main}') old_tips = tuple(row.main for row in previous) closed = tuple(pull for change in changes for pull in change.as_closed()) - Repo.require_commits(reader.checkout, repo.remote, published, *old_tips, - *(pull.head for pull in closed), - *(row.request.head for row in recorded)) + Repo.require_commits(reader.checkout, repo.remote, published, *old_tips) + requested = (*(pull.head for pull in closed), *(row.request.head for row in recorded)) + unservable = Repo.unservable_commits(reader.checkout, repo.remote, *requested) + # A closed head's branch may be gone: fetched or not, no ref reaches it for the graph. + reader.history.include(published, *old_tips, + *(head for head in requested if head not in unservable)) for old in old_tips: if not Repo.descends(reader.checkout, old, published): raise WorkBranchError(f'{repo.github.to_wire()} main rewrote its indexed history') @@ -117,7 +122,7 @@ def observe_repository(store: TodoStore, repo: Repo) -> ClosedRepositoryObservat fingerprints) scan = ClosedHeadScan(store, reader, repo.github, published, frozenset(patch for row in main_rows for patch in row.patches), - fingerprints) + fingerprints, unservable) changed = frozenset(row.number for row in changes) stale = old_tips != (published,) heads = (*(scan.head(row.request, row.closed_at) if stale else row diff --git a/src/agent_comms/commander_service.py b/src/agent_comms/commander_service.py index 605dda423f..88513d2c9b 100644 --- a/src/agent_comms/commander_service.py +++ b/src/agent_comms/commander_service.py @@ -57,13 +57,16 @@ def served(cls, registry: Registration, goals: Goals) -> tuple[Thread, ...]: @classmethod def boards(cls, registry: Registration, goals: Goals) -> tuple[Thread, ...]: - """The registered lead when live and owed service: the one board the heartbeat reads. + """The registered lead's board, the one the heartbeat reads, while any live commander is + owed service: a lead's closed session must not blind the commanders still running. Command moves with the registry, never with an old request: a thread that stood down keeps its parked orders and requests, and gets no heartbeat or vibe check for them.""" snapshot = registry.snapshot() - return tuple( - thread - for thread in snapshot.commander_project.leads(snapshot) - if cls.requested_by(thread, goals) - ) + project = snapshot.commander_project + commanders = project.commanders(snapshot) + if not cls.live(commanders) or not any( + goals.todos.open_requests(thread.acting_identity, limit=1, descending=False) + for thread in commanders): + return () + return project.leads(snapshot) diff --git a/src/agent_comms/repo.py b/src/agent_comms/repo.py index f750c3a90c..b5876bef62 100644 --- a/src/agent_comms/repo.py +++ b/src/agent_comms/repo.py @@ -216,6 +216,18 @@ def require_commits(cls, checkout: Path | str, remote: str, *objects: str) -> No raise UnavailableCommitError(checkout, missing, f"remote {remote}: {error}") from error cls.require_local_commits(checkout, *objects) + @classmethod + def unservable_commits(cls, checkout: Path | str, remote: str, + *objects: str) -> tuple[str, ...]: + """Fetch each missing object alone, so one the remote refuses cannot cost the others; + answer the ones still absent.""" + for name in cls.missing_commits(checkout, *objects): + try: + cls.git(checkout, "fetch", "--quiet", "--no-tags", "--", remote, name) + except GitCommandError: + continue + return cls.missing_commits(checkout, *objects) + def contains(self, sha: str, revision: str = "HEAD") -> bool: """Strict local ancestry; only valid local non-ancestors mean False.""" self.require_local_commits(self.checkout, sha, revision) diff --git a/src/agent_comms/work_branches.py b/src/agent_comms/work_branches.py index daf26363f4..5b6e4dbe8e 100644 --- a/src/agent_comms/work_branches.py +++ b/src/agent_comms/work_branches.py @@ -71,6 +71,17 @@ def description(self, count: int) -> str: return f'has unrelated Git history: {count} commits cannot be compared with main' +@dataclass(frozen=True) +class UnservableCommitBase(CommitBase): + """The forge names a head the remote no longer serves: nothing can be compared with main.""" + + def missing(self, scan: ClosedHeadScan, head: str) -> tuple[str, ...]: + return (head,) + + def description(self, count: int) -> str: + return 'has a head commit origin no longer serves' + + class BranchLink(DeclaredFamily, affix="BranchLink"): """Why a branch's work can be found again.""" diff --git a/tests/test_commander_notifications.py b/tests/test_commander_notifications.py index 157dc25bce..cb7a04ba38 100644 --- a/tests/test_commander_notifications.py +++ b/tests/test_commander_notifications.py @@ -14,7 +14,7 @@ from agent_comms.registry_document import RegisteredCommanderProject -from agent_comms.child_process import ProcessIdentity +from agent_comms.child_process import NoProcess, ProcessIdentity from agent_comms.cli_verbs.fork import ForkCliCommand from agent_comms.thread_management import GivenTags from agent_comms.commander_heartbeat import CommanderHeartbeat, CommanderObservation @@ -888,6 +888,24 @@ def test_a_commander_that_stood_down_keeps_its_requests_but_gets_no_heartbeat(no thread.name for thread in CommanderHeartbeat.boards(comms.registry, comms.goals)} +def test_a_closed_lead_session_leaves_the_board_for_the_live_commanders(notification_bus): + """The lead's process going away must not empty the heartbeat for #commanders still running.""" + comms = notification_bus + comms.registry.register(replace( + comms.registry.require("worker"), tags=frozenset({"comms", CommandersTag.declared_name}), + process_identity=ProcessIdentity.capture(os.getpid()))) + comms.registry.register(replace(comms.registry.require("lead"), + process_identity=NoProcess())) + lead = comms.registry.require("lead") + assert not lead.process_alive + assert CommanderHeartbeat.boards(comms.registry, comms.goals) == (lead,) + beat(comms, (ServicePassFailing(watch="following-rule", since=time.time(), + error="independent source", failed_at=time.time()),)) + worker = comms.registry.require("worker").incarnation + assert CommanderUpdate.latest(comms, worker) + assert CommanderFrame.current(comms, "worker").commander == worker + + def test_one_pass_gives_every_commander_its_own_packet(notification_bus): """#commanders declares who commands: one board read, one packet per live member under its own incarnation. A thread outside the channel, and a fork of a commander (in the channel diff --git a/tests/test_publication_rules.py b/tests/test_publication_rules.py index 1b7c8bc481..c2ceb31ce6 100644 --- a/tests/test_publication_rules.py +++ b/tests/test_publication_rules.py @@ -20,7 +20,7 @@ from agent_comms.recorded_time import RecordedEventTime from agent_comms.task_attachments.commander_pings import RulePing from agent_comms.tool_caller import ThreadCaller -from agent_comms.work_branches import SharedCommitBase, UnrelatedCommitBase +from agent_comms.work_branches import SharedCommitBase, UnrelatedCommitBase, UnservableCommitBase from agent_comms.todos import ( ClosedListedPullRequest, MergedListedPullRequest, NoHeadOrder, ProgressRead, PullRequestCommit, ReviewedHeadVerdict, SupersededHeadVerdict, VerdictReason, @@ -77,6 +77,31 @@ def test_unrelated_closed_head_does_not_poison_its_repository(head_bus, reviewed assert FieldCodec.decode(RulePing, FieldCodec.encode(RulePing.capture(rules[10]))) == RulePing.capture(rules[10]) +def test_closed_head_no_ref_reaches_is_compared_and_an_unservable_one_is_its_own_line( + head_bus, reviewed_source, +): + comms, github, _ = head_bus + source = reviewed_source + stamp = datetime.fromtimestamp(time.time() - 601, UTC).strftime('%Y-%m-%dT%H:%M:%SZ') + head = source.branch_commit + for name in source.git(source.checkout, 'for-each-ref', '--contains', head, + '--format=%(refname)').split(): + source.git(source.checkout, 'update-ref', '-d', name) + gone = 'b' * 40 + for number, closed in ((7, head), (10, gone)): + github.put(source.github, ClosedListedPullRequest( + number, f'closed/{number}', closed, updated=stamp, closed_at=stamp, + base='main', reviewable_times=(stamp,))) + asyncio.run(ClosedPullRequestWatch().run(lambda: nullcontext(comms))) + frame = capture_frame(comms, 'lead', ProgressRead(()), CommanderHeartbeat()) + rules = {rule.request.pr: rule for rule in + ClosedUnmergedPullRequest.scan_observed(comms, 'lead', (), frame)} + assert rules[7].base == SharedCommitBase(source.main_commit) + assert rules[10].base == UnservableCommitBase() + assert rules[10].commits == (gone,) + assert 'has a head commit origin no longer serves' in rules[10].line() + + def test_closed_unmerged_pull_request_flags_lost_commits(head_bus, reviewed_source, installed_runtime): comms, github, _ = head_bus source = reviewed_source