feat(supersede): close + link the contributor PR on land

When a supersede umbrella reaches COMPLETED (our own PR merged), close-on-land
retires the contributor's PR with a linking thank-you comment:

- TaskService.supersede_umbrellas_pending_close() finds landed umbrellas not yet
  marked closed=1; mark_supersede_pr_closed() records the close (idempotent).
- orchestrator._close_superseded_prs runs in the external-PR poll tick: parses
  the contributor PR# from the umbrella's quick_context and calls
  GitService.close_pull_request(delete_branch=False) — we never touch the
  contributor's fork branch. _parse_supersede_pr is unit-tested.

Completes the supersede flow: CEO authorizes -> fork branch -> Main PM -> cell
-> our PR -> CEO merge -> contributor PR closed + linked. ruff + mypy clean
(279); foundation + gateway suites green (5208).

roboco/runtime/orchestrator.py

@@ -4598,9 +4598,55 @@ Start by:
                 )
                 if created is not None:
                     ingested += 1
+        # Close-on-land: retire the contributor PR for any supersede that merged.
+        await self._close_superseded_prs(git, task_service, system_id)
         await db.commit()
         return ingested
 
+    async def _close_superseded_prs(
+        self, git: Any, task_service: Any, system_id: "UUID"
+    ) -> int:
+        """Close + link the contributor PR for each landed supersede umbrella.
+
+        Idempotent: each umbrella is marked ``closed=1`` after its contributor PR
+        is closed, so it is processed once. ``delete_branch=False`` — the
+        contributor's branch lives on their fork; we never touch it. Caller
+        commits.
+        """
+        closed = 0
+        for umbrella in await task_service.supersede_umbrellas_pending_close():
+            pr_number = self._parse_supersede_pr(umbrella.quick_context or "")
+            if pr_number is None:
+                continue
+            try:
+                await git.close_pull_request(
+                    pr_number,
+                    comment=(
+                        "Superseded by the roboco team's own PR — the work was "
+                        "finished and hardened to our standards. Thanks for the "
+                        "contribution!"
+                    ),
+                    delete_branch=False,
+                    actor_agent_id=system_id,
+                )
+            except Exception:
+                logger.exception("close-on-land failed", pr_number=pr_number)
+                continue
+            await task_service.mark_supersede_pr_closed(cast("UUID", umbrella.id))
+            closed += 1
+        return closed
+
+    @staticmethod
+    def _parse_supersede_pr(quick_context: str) -> int | None:
+        """Extract the contributor PR number from a supersede umbrella marker."""
+        for part in quick_context.split():
+            if part.startswith("pr="):
+                try:
+                    return int(part[3:])
+                except ValueError:
+                    return None
+        return None
+
     @staticmethod
     def _pr_author_allowed(pr: dict[str, Any], allowlist: set[str]) -> bool:
         """With a non-empty allowlist, only those GitHub authors are reviewed.

roboco/services/task.py

@@ -821,6 +821,33 @@ class TaskService(BaseService):
                 return task
         return None
 
+    async def supersede_umbrellas_pending_close(self) -> list[TaskTable]:
+        """Landed supersede umbrellas whose contributor PR hasn't been closed yet.
+
+        A supersede umbrella that reached COMPLETED means our own PR merged, so
+        the contributor's PR should be closed + linked. The ``closed=1`` marker
+        in quick_context makes close-on-land idempotent (closed only once).
+        """
+        result = await self.session.execute(
+            select(TaskTable).where(
+                TaskTable.source == "external_pr_supersede",
+                TaskTable.status == TaskStatus.COMPLETED,
+            )
+        )
+        return [
+            task
+            for task in result.scalars().all()
+            if "closed=1" not in (task.quick_context or "")
+        ]
+
+    async def mark_supersede_pr_closed(self, task_id: UUID) -> None:
+        """Record that a landed supersede's contributor PR has been closed."""
+        task = await self.get(task_id)
+        if task is None:
+            return
+        task.quick_context = f"{task.quick_context or ''} closed=1".strip()
+        await self.session.flush()
+
     async def _inherit_parent_session(
         self,
         task_id: UUID,

tests/unit/runtime/test_external_pr_classify.py

@@ -51,3 +51,18 @@ def test_pr_author_allowed(
     pr: dict[str, object], allowlist: set[str], *, expected: bool
 ) -> None:
     assert AgentOrchestrator._pr_author_allowed(pr, allowlist) is expected
+
+
+@pytest.mark.parametrize(
+    ("quick_context", "expected"),
+    [
+        ("external_pr_supersede pr=42 review=abc", 42),
+        ("external_pr_supersede pr=7 review=abc closed=1", 7),
+        ("external_pr_supersede pr=50 review=abc", 50),  # not confused by pr=5
+        ("", None),
+        ("no marker here", None),
+        ("external_pr_supersede pr=notanint review=abc", None),
+    ],
+)
+def test_parse_supersede_pr(quick_context: str, expected: int | None) -> None:
+    assert AgentOrchestrator._parse_supersede_pr(quick_context) == expected