Declare closed heads to the pass graph; keep the board for live commanders

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

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)

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)

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."""
 

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

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