fix(orchestrator): harden external-PR supersede close-on-land

Scope close_pull_request repo resolution by project_id and thread the
umbrella's project into close-on-land, so a contributor PR is never
resolved (or closed) against a same-numbered PR in another project's
repo. Skip the comment + close PATCH when the PR is already closed, so a
retried sweep never re-posts the 'superseded' comment.

Require a non-cancelled descendant that actually landed a PR before
retiring the contributor PR, so an umbrella force-completed over a
cancelled code subtask leaves the contributor's still-valid PR open.

Run close-on-land from the always-on sweeper rather than the default-off
poll loop, so a supersede that lands after the feature is toggled off is
still reconciled. Serialize concurrent supersede triggers under a lock so
a double-click can't cut two branches / spawn two umbrellas. Anchor the
supersede marker checks to the marker line so appended CEO notes can't be
mistaken for the closed/dedup tokens. Make the fork-head branch cut
idempotent (forced refspec) so a commit-fail retry converges.

Also drop an importlib.reload(roboco.config) in a unit test that rebound
the settings singleton and leaked into the PM decision-window test.

roboco/runtime/orchestrator.py

@@ -612,6 +612,11 @@ class AgentOrchestrator:
         self._dispatch_wake: asyncio.Event = asyncio.Event()
         self._running = False
         self._lock = asyncio.Lock()
+        # Serializes CEO supersede calls so a double-click can't pass the
+        # find_supersede_umbrella dedup check twice and cut two branches /
+        # spawn two umbrellas for the same PR (the check is read-then-write
+        # with no DB-level uniqueness).
+        self._supersede_lock = asyncio.Lock()
         # Per-tick set of task_ids already handled by an earlier
         # dispatcher. Reset at the start of every _dispatch_all_work.
         # Consumed via `self._mark_task_handled` / `_is_task_handled`.
@@ -4210,6 +4215,35 @@ Start by:
         # operator's bind-mounted ~/.claude doesn't grow without bound.
         await self._sweep_transcript_retention()
 
+        # Close-on-land for landed supersedes — runs here (always-on sweeper)
+        # rather than the default-off external-PR poll loop, so a supersede that
+        # lands after external_pr_enabled is toggled off is still reconciled.
+        await self._sweep_superseded_prs()
+
+    async def _sweep_superseded_prs(self) -> None:
+        """Retire the contributor PR for any supersede umbrella that landed.
+
+        Dormant in a standard deployment: when no ``external_pr_supersede``
+        umbrellas exist the lookup returns nothing and no GitHub call is made,
+        so this is safe to run unconditionally on every sweep.
+        """
+        from roboco.db.base import get_session_factory
+        from roboco.services.git import GitService
+        from roboco.services.task import get_task_service
+
+        system_id = _foundation.AGENTS["system"].uuid
+        session_factory = get_session_factory()
+        async with session_factory() as db:
+            try:
+                git = GitService(db)
+                task_service = get_task_service(db)
+                closed = await self._close_superseded_prs(git, task_service, system_id)
+                if closed:
+                    await db.commit()
+            except Exception as e:
+                await db.rollback()
+                logger.warning("Supersede close-on-land sweep failed", error=str(e))
+
     async def _sweep_transcript_retention(self) -> None:
         """Prune agent transcripts older than the retention window.
 
@@ -4598,8 +4632,6 @@ 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
 
@@ -4628,9 +4660,16 @@ Start by:
                     ),
                     delete_branch=False,
                     actor_agent_id=system_id,
+                    # PR numbers are per-repo — scope the close to THIS
+                    # umbrella's project so a same-numbered PR in another
+                    # project's repo is never resolved (and closed) by mistake.
+                    project_id=cast("UUID", umbrella.project_id),
                 )
             except Exception:
-                logger.exception("close-on-land failed", pr_number=pr_number)
+                # A permanent close failure (deleted PR, revoked PAT) would
+                # otherwise re-fire + re-log every tick forever; keep it a single
+                # warning rather than a per-tick stack trace.
+                logger.warning("close-on-land failed", pr_number=pr_number)
                 continue
             await task_service.mark_supersede_pr_closed(cast("UUID", umbrella.id))
             closed += 1
@@ -4638,13 +4677,22 @@ Start by:
 
     @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
+        """Extract the contributor PR number from a supersede umbrella marker.
+
+        Anchored to the marker line so a CEO note containing ``pr=`` on a later
+        line of the multi-writer ``quick_context`` can't be misread as the PR.
+        """
+        for raw in quick_context.splitlines():
+            line = raw.strip()
+            if not line.startswith("external_pr_supersede"):
+                continue
+            for part in line.split():
+                if part.startswith("pr="):
+                    try:
+                        return int(part[3:])
+                    except ValueError:
+                        return None
+            return None
         return None
 
     @staticmethod
@@ -4683,7 +4731,9 @@ Start by:
         from roboco.services.project import get_project_service
         from roboco.services.task import get_task_service
 
-        async with get_db_context() as db:
+        # Serialize concurrent CEO calls (double-click) — the dedup check and
+        # the umbrella/branch creation are not atomic across DB sessions.
+        async with self._supersede_lock, get_db_context() as db:
             task_service = get_task_service(db)
             review = await task_service.get(review_task_id)
             if review is None or getattr(review, "source", "") != "external_pr":

roboco/services/git.py

@@ -918,9 +918,13 @@ class GitService(BaseService):
         """
         project_token = await self._token_for_project(project_slug)
         pull_ref = f"refs/pull/{pr_number}/head"
+        # Force the refspec (``+``) so a retry after a prior push (e.g. the
+        # commit after the push failed and rolled the umbrella back) updates the
+        # leftover local branch in the persistent system workspace instead of
+        # hard-erroring on the existing ref — the branch cut is then idempotent.
         await self._run_git(
             workspace,
-            ["fetch", "origin", f"{pull_ref}:{branch_name}"],
+            ["fetch", "origin", f"+{pull_ref}:{branch_name}"],
             token=project_token,
             timeout=_network_git_timeout(),
         )
@@ -2872,20 +2876,29 @@ class GitService(BaseService):
         comment: str | None = None,
         delete_branch: bool = True,
         actor_agent_id: UUID | None = None,
+        project_id: UUID | None = None,
     ) -> None:
         """Close PR ``pr_number`` on GitHub, optionally with an explanatory comment.
 
         Used to retire a PR whose work is already in the base (superseded) so a
         wedged task can complete without a merge — the "close the dead PR"
         action agents had no verb for. Best-effort branch cleanup on close.
+
+        ``pr_number`` alone is ambiguous across projects (GitHub numbers PRs
+        per-repo), so when the caller knows which project the PR belongs to it
+        MUST pass ``project_id`` — the task lookup is then scoped to it so a
+        same-numbered PR in another project's repo is never resolved by
+        accident. Idempotent: a PR that is already closed is a no-op (no
+        duplicate comment), so a retried close-on-land never re-comments.
         """
         from sqlalchemy import select
 
         from roboco.db.tables import TaskTable as _TaskTable
 
-        result = await self.session.execute(
-            select(_TaskTable).where(_TaskTable.pr_number == pr_number).limit(1)
-        )
+        stmt = select(_TaskTable).where(_TaskTable.pr_number == pr_number)
+        if project_id is not None:
+            stmt = stmt.where(_TaskTable.project_id == project_id)
+        result = await self.session.execute(stmt.limit(1))
         task = result.scalar_one_or_none()
         if task is None:
             raise NotFoundError("PR", str(pr_number))
@@ -2904,23 +2917,32 @@ class GitService(BaseService):
             "X-GitHub-Api-Version": "2022-11-28",
         }
         async with httpx.AsyncClient(timeout=_default_git_timeout()) as client:
-            if comment:
-                await client.post(
-                    f"https://api.github.com/repos/{owner}/{repo}/issues/"
-                    f"{pr_number}/comments",
-                    headers=headers,
-                    json={"body": comment},
-                )
-            resp = await client.patch(
+            existing = await client.get(
                 f"https://api.github.com/repos/{owner}/{repo}/pulls/{pr_number}",
                 headers=headers,
-                json={"state": "closed"},
             )
-        if not resp.is_success:
-            raise GitError(
-                f"GitHub API refused PR close ({resp.status_code}): {resp.text[:200]}",
-                {"owner": owner, "repo": repo, "pr": pr_number},
+            already_closed = (
+                existing.is_success and existing.json().get("state") == "closed"
             )
+            if not already_closed:
+                if comment:
+                    await client.post(
+                        f"https://api.github.com/repos/{owner}/{repo}/issues/"
+                        f"{pr_number}/comments",
+                        headers=headers,
+                        json={"body": comment},
+                    )
+                resp = await client.patch(
+                    f"https://api.github.com/repos/{owner}/{repo}/pulls/{pr_number}",
+                    headers=headers,
+                    json={"state": "closed"},
+                )
+                if not resp.is_success:
+                    raise GitError(
+                        f"GitHub API refused PR close ({resp.status_code}): "
+                        f"{resp.text[:200]}",
+                        {"owner": owner, "repo": repo, "pr": pr_number},
+                    )
         if delete_branch:
             await self._delete_pr_branch_best_effort(owner, repo, pr_number, git_token)
 

roboco/services/task.py

@@ -331,6 +331,27 @@ def extract_original_developer(quick_context: str | None) -> str | None:
     return None
 
 
+_SUPERSEDE_MARKER_PREFIX = "external_pr_supersede"
+
+
+def supersede_marker_line(quick_context: str | None) -> str:
+    """Return the supersede marker line from a (multi-writer) quick_context.
+
+    The supersede marker (``external_pr_supersede pr={n} review={uuid}`` plus a
+    ``closed=1`` token once the contributor PR is retired) is always written on
+    its own line, while ``escalate_to_ceo`` / ``ceo_approve`` append free-form
+    CEO notes on later lines. Dedup and close-state checks therefore parse THIS
+    line rather than substring-scanning the whole field — a CEO note that
+    happened to contain ``closed=1`` or ``pr=N review=`` must not be mistaken
+    for the marker (mirrors :func:`extract_original_developer`).
+    """
+    for raw in (quick_context or "").splitlines():
+        line = raw.strip()
+        if line.startswith(_SUPERSEDE_MARKER_PREFIX):
+            return line
+    return ""
+
+
 class TaskService(BaseService):
     """
     Service for managing tasks.
@@ -817,16 +838,20 @@ class TaskService(BaseService):
         )
         needle = f"pr={pr_number} review="
         for task in result.scalars().all():
-            if needle in (task.quick_context or ""):
+            if needle in supersede_marker_line(task.quick_context):
                 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).
+        A supersede umbrella reaching COMPLETED is necessary but not sufficient
+        proof that our replacement PR merged — the CEO can force-complete the
+        root over a *cancelled* code subtask, in which case the team abandoned
+        the work and the contributor's still-valid PR must NOT be retired. So we
+        additionally require a non-cancelled descendant that landed a PR (see
+        :meth:`_supersede_replacement_landed`). The ``closed=1`` token on the
+        marker line makes close-on-land idempotent (closed only once).
         """
         result = await self.session.execute(
             select(TaskTable).where(
@@ -834,18 +859,60 @@ class TaskService(BaseService):
                 TaskTable.status == TaskStatus.COMPLETED,
             )
         )
-        return [
-            task
-            for task in result.scalars().all()
-            if "closed=1" not in (task.quick_context or "")
-        ]
+        pending: list[TaskTable] = []
+        for task in result.scalars().all():
+            if "closed=1" in supersede_marker_line(task.quick_context).split():
+                continue
+            if not await self._supersede_replacement_landed(cast("UUID", task.id)):
+                continue
+            pending.append(task)
+        return pending
+
+    async def _supersede_replacement_landed(self, umbrella_id: UUID) -> bool:
+        """True if a non-cancelled descendant of the umbrella landed a PR.
+
+        Walks the umbrella's subtree (bounded by MAX_TASK_DEPTH) and returns
+        True as soon as it finds a COMPLETED task carrying a ``pr_number`` — the
+        team's merged replacement PR. Returns False when every code descendant
+        was cancelled (force-completed umbrella), so close-on-land then leaves
+        the contributor PR open.
+        """
+        frontier: list[UUID] = [umbrella_id]
+        seen: set[UUID] = set()
+        while frontier:
+            result = await self.session.execute(
+                select(TaskTable).where(TaskTable.parent_task_id.in_(frontier))
+            )
+            frontier = []
+            for child in result.scalars().all():
+                child_id = cast("UUID", child.id)
+                if child_id in seen:
+                    continue
+                seen.add(child_id)
+                if child.status == TaskStatus.COMPLETED and child.pr_number is not None:
+                    return True
+                frontier.append(child_id)
+        return False
 
     async def mark_supersede_pr_closed(self, task_id: UUID) -> None:
-        """Record that a landed supersede's contributor PR has been closed."""
+        """Record that a landed supersede's contributor PR has been closed.
+
+        Appends ``closed=1`` to the marker LINE (not the end of the whole
+        multi-writer field) so the idempotency token stays anchored to the
+        marker and survives appended CEO notes.
+        """
         task = await self.get(task_id)
         if task is None:
             return
-        task.quick_context = f"{task.quick_context or ''} closed=1".strip()
+        lines = (task.quick_context or "").splitlines()
+        for i, raw in enumerate(lines):
+            if raw.strip().startswith(_SUPERSEDE_MARKER_PREFIX):
+                if "closed=1" not in raw.split():
+                    lines[i] = f"{raw} closed=1"
+                break
+        else:
+            lines.append(f"{_SUPERSEDE_MARKER_PREFIX} closed=1")
+        task.quick_context = "\n".join(lines)
         await self.session.flush()
 
     async def _inherit_parent_session(

tests/unit/runtime/test_external_pr_classify.py

@@ -62,6 +62,8 @@ def test_pr_author_allowed(
         ("", None),
         ("no marker here", None),
         ("external_pr_supersede pr=notanint review=abc", None),
+        # A CEO note on a later line must not shadow the marker line's pr=.
+        ("external_pr_supersede pr=11 review=abc\nceo_approval_notes: pr=99 ok", 11),
     ],
 )
 def test_parse_supersede_pr(quick_context: str, expected: int | None) -> None:

tests/unit/services/test_git_commit_trailer.py

@@ -1,7 +1,5 @@
 """Unit tests: commit-trailer links use ROBOCO_PUBLIC_BASE_URL."""
 
-import importlib
-
 import pytest
 import roboco.config as config_module
 from roboco.templates.git.commit import (
@@ -41,17 +39,18 @@ def test_build_commit_message_strips_trailing_slash() -> None:
 
 def test_links_use_public_base_url(monkeypatch: pytest.MonkeyPatch) -> None:
     """settings.public_base_url drives the api_base passed to build_commit_message."""
+    # Construct a fresh Settings() to read the patched env — do NOT
+    # importlib.reload(roboco.config), which would rebind the module-level
+    # ``settings`` singleton to a new object and leak that divergence into any
+    # other test that captured the original reference via ``from roboco.config
+    # import settings`` (e.g. the PM decision-window gate).
     monkeypatch.setenv("ROBOCO_PUBLIC_BASE_URL", "https://roboco.example.com")
-    importlib.reload(config_module)
-    try:
-        settings = config_module.Settings()
-        api_base = settings.public_base_url.rstrip("/") + "/api"
-        ctx = _make_ctx()
-        out = build_commit_message(ctx, api_base)
-        assert "https://roboco.example.com" in out
-        assert "127.0.0.1" not in out
-    finally:
-        importlib.reload(config_module)
+    settings = config_module.Settings()
+    api_base = settings.public_base_url.rstrip("/") + "/api"
+    ctx = _make_ctx()
+    out = build_commit_message(ctx, api_base)
+    assert "https://roboco.example.com" in out
+    assert "127.0.0.1" not in out
 
 
 def test_commit_context_invalid_type_raises() -> None:

tests/unit/services/test_git_rebase_resolve.py

@@ -161,6 +161,9 @@ async def test_close_pull_request_patches_state_closed(
         status_code = 200
         text = ""
 
+        def json(self) -> dict[str, str]:
+            return {"state": "open"}
+
     class _Client:
         async def __aenter__(self) -> _Client:
             return self
@@ -168,6 +171,10 @@ async def test_close_pull_request_patches_state_closed(
         async def __aexit__(self, *_a: Any) -> None:
             return None
 
+        async def get(self, url: str, **_kw: Any) -> _Resp:
+            calls.append(("GET", url))
+            return _Resp()
+
         async def post(self, url: str, **_kw: Any) -> _Resp:
             calls.append(("POST", url))
             return _Resp()
@@ -185,3 +192,74 @@ async def test_close_pull_request_patches_state_closed(
     ) in calls
     assert ("PATCH", "https://api.github.com/repos/owner/repo/pulls/159") in calls
     delete_branch.assert_awaited_once()
+
+
+@pytest.mark.asyncio
+async def test_close_pull_request_idempotent_when_already_closed(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """An already-closed PR is a no-op: no duplicate comment, no PATCH.
+
+    Guards the close-on-land retry path — a transient failure between the
+    comment POST and the close PATCH must not re-post the explanatory comment
+    on the next sweep.
+    """
+    svc = _git_service()
+    task = type("T", (), {"id": "t", "assigned_to": None, "created_by": None})()
+    session = AsyncMock()
+    session.execute = AsyncMock(
+        return_value=type("Res", (), {"scalar_one_or_none": lambda _self: task})()
+    )
+    monkeypatch.setattr(svc, "session", session, raising=False)
+    monkeypatch.setattr(
+        svc,
+        "_project_for_task",
+        AsyncMock(return_value=type("P", (), {"slug": "proj"})()),
+    )
+    monkeypatch.setattr(
+        svc, "_resolve_workspace_agent_id", MagicMock(return_value=None)
+    )
+    monkeypatch.setattr(svc, "get_workspace", AsyncMock(return_value=Path("/tmp/ws")))
+    monkeypatch.setattr(
+        svc, "_get_project_token_or_raise", AsyncMock(return_value="tok")
+    )
+    monkeypatch.setattr(
+        svc, "_parse_github_remote", MagicMock(return_value=("owner", "repo"))
+    )
+    monkeypatch.setattr(svc, "_delete_pr_branch_best_effort", AsyncMock())
+
+    calls: list[tuple[str, str]] = []
+
+    class _Resp:
+        is_success = True
+        status_code = 200
+        text = ""
+
+        def json(self) -> dict[str, str]:
+            return {"state": "closed"}
+
+    class _Client:
+        async def __aenter__(self) -> _Client:
+            return self
+
+        async def __aexit__(self, *_a: Any) -> None:
+            return None
+
+        async def get(self, url: str, **_kw: Any) -> _Resp:
+            calls.append(("GET", url))
+            return _Resp()
+
+        async def post(self, url: str, **_kw: Any) -> _Resp:
+            calls.append(("POST", url))
+            return _Resp()
+
+        async def patch(self, url: str, **_kw: Any) -> _Resp:
+            calls.append(("PATCH", url))
+            return _Resp()
+
+    with patch("roboco.services.git.httpx.AsyncClient", return_value=_Client()):
+        await svc.close_pull_request(
+            159, comment="superseded by #158", delete_branch=False
+        )
+
+    assert [c[0] for c in calls] == ["GET"]  # no POST comment, no PATCH

tests/unit/services/test_supersede_umbrella.py

@@ -0,0 +1,171 @@
+"""Supersede umbrella close-on-land guards.
+
+Covers the parts of the external-PR supersede flow that decide whether — and
+which — a landed supersede's contributor PR gets retired:
+
+- ``supersede_marker_line`` anchors marker/state checks to the marker line, so
+  free-form CEO escalation/approval notes appended to the same multi-writer
+  ``quick_context`` can't be mistaken for the marker.
+- ``supersede_umbrellas_pending_close`` only returns umbrellas whose
+  replacement work actually landed (a non-cancelled descendant carrying a PR),
+  not every COMPLETED umbrella — the CEO can force-complete over a cancelled
+  code subtask.
+- ``mark_supersede_pr_closed`` writes the ``closed=1`` idempotency token onto
+  the marker line, surviving appended notes.
+"""
+
+from __future__ import annotations
+
+from unittest.mock import AsyncMock, MagicMock
+from uuid import uuid4
+
+import pytest
+from roboco.models.base import TaskStatus
+from roboco.services.task import TaskService, supersede_marker_line
+
+_MARKER = "external_pr_supersede pr=5 review=abc"
+
+
+def _scalars_all(rows: list[object]) -> MagicMock:
+    """A session.execute return value whose .scalars().all() yields `rows`."""
+    res = MagicMock()
+    res.scalars.return_value.all.return_value = rows
+    return res
+
+
+def _service(execute_returns: object) -> TaskService:
+    session = MagicMock()
+    session.execute = AsyncMock(return_value=execute_returns)
+    session.flush = AsyncMock()
+    return TaskService(session)
+
+
+def _bind(svc: TaskService, name: str, value: object) -> None:
+    object.__setattr__(svc, name, value)
+
+
+# ---------------------------------------------------------------------------
+# supersede_marker_line — line anchoring
+# ---------------------------------------------------------------------------
+
+
+def test_marker_line_returns_marker_ignoring_appended_notes() -> None:
+    qc = f"{_MARKER}\nceo_approval_notes: shipped, looks good"
+    assert supersede_marker_line(qc) == _MARKER
+
+
+def test_marker_line_not_fooled_by_closed_token_in_note() -> None:
+    qc = f"{_MARKER}\nceo_approval_notes: marked closed=1 in jira"
+    # The marker line itself carries no closed=1, so the PR is NOT yet closed.
+    assert "closed=1" not in supersede_marker_line(qc).split()
+
+
+def test_marker_line_empty_when_absent() -> None:
+    assert supersede_marker_line("no marker here\nescalation_notes: x") == ""
+    assert supersede_marker_line(None) == ""
+
+
+# ---------------------------------------------------------------------------
+# supersede_umbrellas_pending_close — closed-token + landed-replacement gates
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_pending_close_excludes_umbrella_with_closed_marker() -> None:
+    umbrella = MagicMock(id=uuid4(), quick_context=f"{_MARKER} closed=1")
+    svc = _service(_scalars_all([umbrella]))
+    _bind(svc, "_supersede_replacement_landed", AsyncMock(return_value=True))
+    assert await svc.supersede_umbrellas_pending_close() == []
+
+
+@pytest.mark.asyncio
+async def test_pending_close_keeps_umbrella_with_closed_token_only_in_note() -> None:
+    # A CEO note containing the literal "closed=1" must NOT retire the PR.
+    umbrella = MagicMock(
+        id=uuid4(), quick_context=f"{_MARKER}\nceo_approval_notes: closed=1 elsewhere"
+    )
+    svc = _service(_scalars_all([umbrella]))
+    _bind(svc, "_supersede_replacement_landed", AsyncMock(return_value=True))
+    out = await svc.supersede_umbrellas_pending_close()
+    assert out == [umbrella]
+
+
+@pytest.mark.asyncio
+async def test_pending_close_requires_landed_replacement() -> None:
+    # COMPLETED + no closed marker, but the replacement never landed (the code
+    # subtask was cancelled) — close-on-land must skip it.
+    umbrella = MagicMock(id=uuid4(), quick_context=_MARKER)
+    svc = _service(_scalars_all([umbrella]))
+    _bind(svc, "_supersede_replacement_landed", AsyncMock(return_value=False))
+    assert await svc.supersede_umbrellas_pending_close() == []
+
+
+# ---------------------------------------------------------------------------
+# _supersede_replacement_landed — subtree walk
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_replacement_landed_true_for_completed_descendant_with_pr() -> None:
+    child = MagicMock(id=uuid4(), status=TaskStatus.COMPLETED, pr_number=42)
+    svc = _service(_scalars_all([child]))
+    assert await svc._supersede_replacement_landed(uuid4()) is True
+
+
+@pytest.mark.asyncio
+async def test_replacement_landed_false_when_descendant_cancelled() -> None:
+    child = MagicMock(id=uuid4(), status=TaskStatus.CANCELLED, pr_number=42)
+    svc = _service(_scalars_all([child]))
+    # The `seen` guard terminates the walk even though the mock re-returns child.
+    assert await svc._supersede_replacement_landed(uuid4()) is False
+
+
+@pytest.mark.asyncio
+async def test_replacement_landed_false_when_completed_without_pr() -> None:
+    child = MagicMock(id=uuid4(), status=TaskStatus.COMPLETED, pr_number=None)
+    svc = _service(_scalars_all([child]))
+    assert await svc._supersede_replacement_landed(uuid4()) is False
+
+
+# ---------------------------------------------------------------------------
+# find_supersede_umbrella — marker-line dedup
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_find_umbrella_matches_marker_not_note() -> None:
+    match = MagicMock(id=uuid4(), quick_context=f"{_MARKER}\nescalation_notes: x")
+    other = MagicMock(
+        id=uuid4(),
+        # marker for a different PR, but a note mentions "pr=5 review=" text
+        quick_context="external_pr_supersede pr=9 review=z\nnote: see pr=5 review= ok",
+    )
+    svc = _service(_scalars_all([other, match]))
+    found = await svc.find_supersede_umbrella(uuid4(), 5)
+    assert found is match
+
+
+# ---------------------------------------------------------------------------
+# mark_supersede_pr_closed — token written on the marker line
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_mark_closed_appends_token_to_marker_line() -> None:
+    task = MagicMock(quick_context=f"{_MARKER}\nceo_approval_notes: shipped")
+    svc = _service(_scalars_all([]))
+    _bind(svc, "get", AsyncMock(return_value=task))
+    await svc.mark_supersede_pr_closed(uuid4())
+    lines = task.quick_context.splitlines()
+    assert lines[0] == f"{_MARKER} closed=1"
+    assert lines[1] == "ceo_approval_notes: shipped"  # note untouched
+
+
+@pytest.mark.asyncio
+async def test_mark_closed_is_idempotent_on_marker_line() -> None:
+    task = MagicMock(quick_context=f"{_MARKER} closed=1\nceo_approval_notes: x")
+    svc = _service(_scalars_all([]))
+    _bind(svc, "get", AsyncMock(return_value=task))
+    await svc.mark_supersede_pr_closed(uuid4())
+    # No second closed=1 token appended.
+    assert task.quick_context.splitlines()[0] == f"{_MARKER} closed=1"