From 9b34db9e759a325a53724dbb1084b70dbc5ecced Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Fri, 30 Oct 2026 18:07:18 -0400 Subject: [PATCH] Declare closed heads to the pass's graph; keep the board for live commanders ClosedPullRequestWatch fetched each closed head but never included it in CheckoutHistory, whose graph holds only what a ref reaches. Once a closed PR's branch ref is pruned, repo.py:902 calls a commit that is on disk "unavailable" and the pass fails for good, which is why fetching it by hand cleared nothing. Every head now enters the graph, and a head the remote refuses is its own UnservableCommitBase line instead of killing the pass. CommanderService.boards returned nothing when the registered lead's process was gone, so the pass completed nobody and replace_all deleted every live commander's frame. The lead's board now stands while any commander is live and any commander is owed service. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01XbFMSpxUkNUcAPqceQ1zkG --- src/agent_comms/closed_pull_requests.py | 17 ++++++++++------ src/agent_comms/commander_service.py | 15 ++++++++------ src/agent_comms/repo.py | 12 +++++++++++ src/agent_comms/work_branches.py | 11 ++++++++++ tests/test_commander_notifications.py | 20 +++++++++++++++++- tests/test_publication_rules.py | 27 ++++++++++++++++++++++++- 6 files changed, 88 insertions(+), 14 deletions(-) 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