fix(gateway): gate review diffs against the task's real parent branch (#444) (#454)

The in-path PR-review gate's evidence diff (claim_gate_review) and the
pr_pass conventions guard derived their diff base via parent_branch_for
string surgery, which reuses the child branch's own team segment — wrong
for every cross-team hop (a frontend child of a main_pm root derives a
ref that never existed) and silently falls back to the repo default
branch, so the reviewer judged the entire inherited base-branch content
as the task's own work and failed acceptance criteria the task never
touched. Bounced a live goals-tab fix three times, unfixable by branch
surgery.

The gate now resolves the base via resolve_parent_branch (the parent
task's recorded branch_name, cross-team correct) and threads it as a new
preferred_parent override through git.diff / list_changed_files /
conventions_check_for_task — consulted only when no explicit base is
given, so the pinned literal-base contract (base="HEAD~1") and every
other diff caller (QA, doc, content) are byte-identical. Parent lookup
fails open (derived-base fallback) like the other resolve_parent_branch
call sites, and is skipped entirely while the conventions flag is off.

Also excludes .uv-cache/ and .claude/ (agent worktrees, private uv
cache) from the markdown prose scanner — both are repo-local tool dirs
whose vendored/generated files tripped make reflow-check.

Co-authored-by: Renn F <rennf93@users.noreply.github.com>

roboco/services/gateway/choreographer/_impl.py

@@ -2215,7 +2215,11 @@ class Choreographer:
             )
 
     async def _conventions_guard(
-        self, agent_id: UUID, task: Any, briefing: dict[str, Any]
+        self,
+        agent_id: UUID,
+        task: Any,
+        briefing: dict[str, Any],
+        preferred_parent: str | None = None,
     ) -> Envelope | None:
         """Run the conventions validator on the actor's changed files (gated).
 
@@ -2225,12 +2229,19 @@ class Choreographer:
         flag is off. This is the pr_pass (reviewer) path — the remediation is
         reviewer-aware (``pr_fail``, not ``i_am_blocked`` which a reviewer
         lacks) via ``_conventions_rejection(..., reviewer=True)``.
+
+        ``preferred_parent`` is the assembled task's real parent branch (see
+        ``PRGateMixin._gate_diff_parent``) — the same cross-team-correct base
+        the gate's own diff evidence uses, so a misplaced-definition finding
+        can't be raised against inherited base-branch content.
         """
         from roboco.config import settings as _settings
 
         if not _settings.conventions_enabled:
             return None
-        result = await self.git.conventions_check_for_task(agent_id, task)
+        result = await self.git.conventions_check_for_task(
+            agent_id, task, preferred_parent=preferred_parent
+        )
         return self._conventions_rejection(result, briefing, reviewer=True)
 
     @staticmethod

roboco/services/gateway/choreographer/_protocol.py

@@ -62,7 +62,11 @@ class ChoreographerHelpers:
         raise NotImplementedError
 
     async def _conventions_guard(
-        self, agent_id: UUID, task: Any, briefing: dict[str, Any]
+        self,
+        agent_id: UUID,
+        task: Any,
+        briefing: dict[str, Any],
+        preferred_parent: str | None = None,
     ) -> Envelope | None:
         raise NotImplementedError
 

roboco/services/gateway/choreographer/pr_gate.py

@@ -25,6 +25,7 @@ from roboco.foundation.policy import tracing as _tr
 from roboco.foundation.policy.batch import is_batch_root_subtask
 from roboco.foundation.policy.content import markers
 from roboco.services.gateway.envelope import Envelope
+from roboco.services.gateway.merge_chain import resolve_parent_branch
 
 if TYPE_CHECKING:
     from uuid import UUID
@@ -440,9 +441,18 @@ class PRGateMixin(_Base):
         violations; pr_fail stays available. Returns the emitted rejection or
         None to proceed. Both guards are inert when their flag is off.
         """
+        from roboco.config import settings as _settings
+
+        # Only the conventions guard consumes the parent — skip the lookup
+        # entirely (and its failure surface) while the flag is off.
+        parent = (
+            await self._gate_diff_parent(t) if _settings.conventions_enabled else None
+        )
         guards = (
             lambda: self._toolchain_broken_guard(reviewer_agent_id, t, reviewer=True),
-            lambda: self._conventions_guard(reviewer_agent_id, t, briefing),
+            lambda: self._conventions_guard(
+                reviewer_agent_id, t, briefing, preferred_parent=parent
+            ),
         )
         for guard in guards:
             rejection = await guard()
@@ -696,11 +706,34 @@ class PRGateMixin(_Base):
             verb=verb,
         )
 
+    async def _gate_diff_parent(self, t: Any) -> str | None:
+        """The assembled task's real parent branch, or None (branchless task).
+
+        ``resolve_parent_branch`` reads the parent TASK's own ``branch_name``
+        (correct across a team boundary — every cell→root hop, where the
+        child's own team segment can't derive the root's ``main_pm``
+        branch), unlike the string-derived ``parent_branch_for`` that
+        ``git.diff``'s default base falls back on. Fail-open on a lookup
+        error (None → the derived-base fallback), like every other
+        ``resolve_parent_branch`` call site — a transient DB miss degrades
+        the diff base, never 500s the gate verb.
+        """
+        if not t.branch_name:
+            return None
+        try:
+            return await resolve_parent_branch(t, self.task)
+        except Exception as exc:
+            logger.warning("gate_diff_parent_skip", task_id=str(t.id), error=str(exc))
+            return None
+
     async def _build_gate_review_evidence(self, t: Any) -> dict[str, Any]:
         """Inline evidence for claim_gate_review: the assembled diff + criteria."""
         diff = ""
         if t.branch_name:
-            diff = await self.git.diff(branch_name=t.branch_name)
+            diff = await self.git.diff(
+                branch_name=t.branch_name,
+                preferred_parent=await self._gate_diff_parent(t),
+            )
         return {
             "pr_number": t.pr_number,
             "pr_url": t.pr_url,

roboco/services/git.py

@@ -4550,7 +4550,12 @@ class GitService(BaseService):
         return "origin/master"
 
     async def _resolve_diff_base(
-        self, workspace: Any, branch_name: str, token: str | None = None
+        self,
+        workspace: Any,
+        branch_name: str,
+        token: str | None = None,
+        *,
+        preferred_parent: str | None = None,
     ) -> str:
         """Best diff base for `branch_name` when no explicit base is given.
 
@@ -4567,10 +4572,23 @@ class GitService(BaseService):
         a stale base spans the whole repo delta, not the branch's change.
         Re-fetch the resolved base authenticated (unauth fails on private
         repos) so the base is current.
+
+        ``preferred_parent``, when given, overrides the string-derived
+        ``parent_branch_for`` with an authoritative parent branch name (e.g.
+        ``merge_chain.resolve_parent_branch``, which reads the parent TASK's
+        own ``branch_name`` — correct across a team boundary, unlike the
+        derivation below which reuses ``branch_name``'s own team segment).
+        Still falls back to the repo default branch when that parent was
+        never pushed, so an unassembled/branchless parent can't crash the
+        diff.
         """
         from roboco.services.gateway.merge_chain import parent_branch_for
 
-        parent = parent_branch_for(branch_name)
+        parent = (
+            preferred_parent
+            if preferred_parent is not None
+            else parent_branch_for(branch_name)
+        )
         await self._run_git(
             workspace, ["fetch", "origin", parent], check=False, token=token
         )
@@ -4632,6 +4650,7 @@ class GitService(BaseService):
         branch_name: str,
         base: str | None = None,
         actor_agent_id: UUID | None = None,
+        preferred_parent: str | None = None,
     ) -> str:
         """Return the git diff for `branch_name` against `base`.
 
@@ -4643,6 +4662,11 @@ class GitService(BaseService):
         ``actor_agent_id`` resolves the workspace via the caller's clone
         when ``task.assigned_to`` is None — important for
         QA reviewing post-submit_qa.
+
+        ``preferred_parent`` is ignored once ``base`` is explicit; it only
+        overrides the derived-parent lookup (see ``_resolve_diff_base``) for
+        a caller with an authoritative parent branch name (a cross-team
+        assembled-PR review) — never a literal ref like ``base="HEAD~1"``.
         """
         workspace = await self._workspace_for_branch(
             branch_name, actor_agent_id=actor_agent_id
@@ -4652,7 +4676,9 @@ class GitService(BaseService):
         base_ref = (
             base
             if base is not None
-            else await self._resolve_diff_base(workspace, branch_name, token=token)
+            else await self._resolve_diff_base(
+                workspace, branch_name, token=token, preferred_parent=preferred_parent
+            )
         )
         diff_result = await self._run_git(
             workspace, ["diff", f"{base_ref}...{head_ref}"], check=False
@@ -4665,6 +4691,7 @@ class GitService(BaseService):
         branch_name: str,
         base: str | None = None,
         actor_agent_id: UUID | None = None,
+        preferred_parent: str | None = None,
     ) -> list[str]:
         """Return the file paths changed on `branch_name` relative to `base`.
 
@@ -4674,7 +4701,7 @@ class GitService(BaseService):
         ever called the legacy ``add_files_modified`` HTTP endpoint
         (which the gateway commit() does not call). Empty paths are
         skipped; output preserves git's order. Same default-
-        branch fallback as ``diff``.
+        branch fallback as ``diff`` (including ``preferred_parent``).
         """
         workspace = await self._workspace_for_branch(
             branch_name, actor_agent_id=actor_agent_id
@@ -4684,7 +4711,9 @@ class GitService(BaseService):
         base_ref = (
             base
             if base is not None
-            else await self._resolve_diff_base(workspace, branch_name, token=token)
+            else await self._resolve_diff_base(
+                workspace, branch_name, token=token, preferred_parent=preferred_parent
+            )
         )
         result = await self._run_git(
             workspace,
@@ -4803,7 +4832,11 @@ class GitService(BaseService):
         }
 
     async def conventions_check_for_task(
-        self, actor_agent_id: UUID | None, task: Any
+        self,
+        actor_agent_id: UUID | None,
+        task: Any,
+        *,
+        preferred_parent: str | None = None,
     ) -> dict[str, Any]:
         """Run the conventions validator on a task's changed files.
 
@@ -4815,6 +4848,9 @@ class GitService(BaseService):
         exit-3 philosophy). The two empty-result paths stay fail-open: a
         branchless task (no ``branch_name``) and a task with no changed files
         genuinely have nothing to validate, so the gate correctly passes.
+
+        ``preferred_parent`` threads to ``list_changed_files`` — the in-path
+        PR-review gate's cross-team parent (see ``diff``'s docstring).
         """
         try:
             branch = task.branch_name
@@ -4824,7 +4860,9 @@ class GitService(BaseService):
                 branch, actor_agent_id=actor_agent_id
             )
             changed = await self.list_changed_files(
-                branch_name=branch, actor_agent_id=actor_agent_id
+                branch_name=branch,
+                actor_agent_id=actor_agent_id,
+                preferred_parent=preferred_parent,
             )
         except Exception as exc:
             return {

scripts/reflow_md.py

@@ -28,6 +28,8 @@ SKIP_DIRS = {
     ".OLD",
     "node_modules",
     ".venv",
+    ".uv-cache",
+    ".claude",
     ".next",
     "dist",
     ".git",

tests/unit/gateway/test_gate_review_diff_base.py

@@ -0,0 +1,227 @@
+"""In-path PR-review gate: the assembled diff must use the REAL parent branch.
+
+``_build_gate_review_evidence`` (claim_gate_review) and ``_pr_pass_blocked``
+(pr_pass's conventions guard) used to call ``git.diff`` / the conventions
+check with no base, which derives the parent via the same-team string
+surgery ``parent_branch_for`` — wrong for every cross-team cell→root hop
+(the cell task's own team segment can't derive the ``main_pm`` root's
+branch). Both now resolve ``preferred_parent`` via
+``merge_chain.resolve_parent_branch`` (reads the parent TASK's own
+``branch_name``) and thread it through, falling back exactly like the
+pre-fix derivation for a root / branchless-parent / parentless task.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+from unittest.mock import AsyncMock, MagicMock
+from uuid import uuid4
+
+import pytest
+from roboco.config import settings
+from roboco.services.gateway.choreographer import Choreographer, ChoreographerDeps
+
+
+def _make_choreographer(*, task_service: AsyncMock, git: AsyncMock) -> Choreographer:
+    return Choreographer(
+        ChoreographerDeps(
+            task=task_service,
+            work_session=AsyncMock(),
+            git=git,
+            a2a=AsyncMock(),
+            journal=AsyncMock(),
+            audit=AsyncMock(),
+            evidence_repo=AsyncMock(),
+        )
+    )
+
+
+def _gate_task(*, branch_name: str, parent_task_id: Any) -> Any:
+    return MagicMock(
+        branch_name=branch_name,
+        parent_task_id=parent_task_id,
+        pr_number=139,
+        pr_url="https://example/pr/139",
+        acceptance_criteria=[],
+    )
+
+
+class TestGateDiffParent:
+    """``_gate_diff_parent`` mirrors ``resolve_parent_branch``'s three cases."""
+
+    @pytest.mark.asyncio
+    async def test_cross_team_child_uses_parent_task_branch(self) -> None:
+        parent_id = uuid4()
+        t = _gate_task(
+            branch_name="feature/frontend/f7d0a61a--e56e6543--e2b50b06",
+            parent_task_id=parent_id,
+        )
+        task_service = AsyncMock()
+        task_service.get.return_value = MagicMock(
+            branch_name="feature/main_pm/f7d0a61a--e56e6543"
+        )
+        c = _make_choreographer(task_service=task_service, git=AsyncMock())
+
+        parent = await c._gate_diff_parent(t)
+        assert parent == "feature/main_pm/f7d0a61a--e56e6543"
+        task_service.get.assert_awaited_once_with(parent_id)
+
+    @pytest.mark.asyncio
+    async def test_root_subtask_with_branchless_umbrella_uses_project_default(
+        self,
+    ) -> None:
+        parent_id = uuid4()
+        t = _gate_task(
+            branch_name="feature/main_pm/f7d0a61a--e56e6543", parent_task_id=parent_id
+        )
+        task_service = AsyncMock()
+        task_service.get.return_value = MagicMock(branch_name=None)
+        task_service.project_default_branch_for_task = AsyncMock(return_value="master")
+        c = _make_choreographer(task_service=task_service, git=AsyncMock())
+
+        parent = await c._gate_diff_parent(t)
+        assert parent == "master"
+
+    @pytest.mark.asyncio
+    async def test_parentless_root_falls_back_to_string_derivation(self) -> None:
+        t = _gate_task(branch_name="feature/main_pm/f7d0a61a", parent_task_id=None)
+        task_service = AsyncMock()
+        c = _make_choreographer(task_service=task_service, git=AsyncMock())
+
+        parent = await c._gate_diff_parent(t)
+        assert parent == "master"
+        task_service.get.assert_not_called()
+
+    @pytest.mark.asyncio
+    async def test_branchless_task_returns_none(self) -> None:
+        t = _gate_task(branch_name="", parent_task_id=uuid4())
+        task_service = AsyncMock()
+        c = _make_choreographer(task_service=task_service, git=AsyncMock())
+
+        assert await c._gate_diff_parent(t) is None
+        task_service.get.assert_not_called()
+
+    @pytest.mark.asyncio
+    async def test_fails_open_on_parent_lookup_error(self) -> None:
+        t = _gate_task(
+            branch_name="feature/frontend/f7d0a61a--e56e6543--e2b50b06",
+            parent_task_id=uuid4(),
+        )
+        task_service = AsyncMock()
+        task_service.get.side_effect = RuntimeError("db connection reset")
+        c = _make_choreographer(task_service=task_service, git=AsyncMock())
+
+        assert await c._gate_diff_parent(t) is None
+
+
+class TestBuildGateReviewEvidence:
+    @pytest.mark.asyncio
+    async def test_diff_called_with_resolved_cross_team_parent(self) -> None:
+        parent_id = uuid4()
+        t = _gate_task(
+            branch_name="feature/frontend/f7d0a61a--e56e6543--e2b50b06",
+            parent_task_id=parent_id,
+        )
+        task_service = AsyncMock()
+        task_service.get.return_value = MagicMock(
+            branch_name="feature/main_pm/f7d0a61a--e56e6543"
+        )
+        git = AsyncMock()
+        git.diff.return_value = "diff body"
+        c = _make_choreographer(task_service=task_service, git=git)
+
+        evidence = await c._build_gate_review_evidence(t)
+
+        git.diff.assert_awaited_once_with(
+            branch_name=t.branch_name,
+            preferred_parent="feature/main_pm/f7d0a61a--e56e6543",
+        )
+        assert evidence["pr_diff"] == "diff body"
+
+    @pytest.mark.asyncio
+    async def test_diff_skipped_for_branchless_task(self) -> None:
+        t = _gate_task(branch_name="", parent_task_id=None)
+        git = AsyncMock()
+        c = _make_choreographer(task_service=AsyncMock(), git=git)
+
+        evidence = await c._build_gate_review_evidence(t)
+
+        git.diff.assert_not_awaited()
+        assert evidence["pr_diff"] == ""
+
+    @pytest.mark.asyncio
+    async def test_diff_falls_back_when_parent_lookup_fails(self) -> None:
+        t = _gate_task(
+            branch_name="feature/frontend/f7d0a61a--e56e6543--e2b50b06",
+            parent_task_id=uuid4(),
+        )
+        task_service = AsyncMock()
+        task_service.get.side_effect = RuntimeError("db connection reset")
+        git = AsyncMock()
+        git.diff.return_value = "diff body"
+        c = _make_choreographer(task_service=task_service, git=git)
+
+        evidence = await c._build_gate_review_evidence(t)
+
+        git.diff.assert_awaited_once_with(
+            branch_name=t.branch_name, preferred_parent=None
+        )
+        assert evidence["pr_diff"] == "diff body"
+
+
+class TestPrPassBlockedThreadsParent:
+    """``_pr_pass_blocked`` resolves the parent ONCE and hands it to the
+    conventions guard, so a reviewer's block-level finding is never raised
+    against inherited base-branch content on a cross-team assembled PR."""
+
+    @pytest.mark.asyncio
+    async def test_conventions_guard_receives_resolved_parent(
+        self, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        monkeypatch.setattr(settings, "conventions_enabled", True)
+        parent_id = uuid4()
+        t = _gate_task(
+            branch_name="feature/frontend/f7d0a61a--e56e6543--e2b50b06",
+            parent_task_id=parent_id,
+        )
+        task_service = AsyncMock()
+        task_service.get.return_value = MagicMock(
+            branch_name="feature/main_pm/f7d0a61a--e56e6543"
+        )
+        c = _make_choreographer(task_service=task_service, git=AsyncMock())
+        cc: Any = c
+        cc._toolchain_broken_guard = AsyncMock(return_value=None)
+        cc._conventions_guard = AsyncMock(return_value=None)
+        reviewer_id = uuid4()
+
+        result = await c._pr_pass_blocked(reviewer_id, uuid4(), t, "pr_reviewer", {})
+
+        assert result is None
+        cc._conventions_guard.assert_awaited_once_with(
+            reviewer_id,
+            t,
+            {},
+            preferred_parent="feature/main_pm/f7d0a61a--e56e6543",
+        )
+
+    @pytest.mark.asyncio
+    async def test_parent_lookup_skipped_when_conventions_off(
+        self, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        monkeypatch.setattr(settings, "conventions_enabled", False)
+        t = _gate_task(
+            branch_name="feature/frontend/f7d0a61a--e56e6543--e2b50b06",
+            parent_task_id=uuid4(),
+        )
+        task_service = AsyncMock()
+        c = _make_choreographer(task_service=task_service, git=AsyncMock())
+        cc: Any = c
+        cc._toolchain_broken_guard = AsyncMock(return_value=None)
+        cc._conventions_guard = AsyncMock(return_value=None)
+
+        result = await c._pr_pass_blocked(uuid4(), uuid4(), t, "pr_reviewer", {})
+
+        assert result is None
+        task_service.get.assert_not_called()
+        cc._conventions_guard.assert_awaited_once()
+        assert cc._conventions_guard.await_args.kwargs.get("preferred_parent") is None

tests/unit/services/test_git_conventions_check_fail_closed.py

@@ -95,6 +95,28 @@ async def test_no_changed_files_still_fails_open() -> None:
     assert result["findings"] == []
 
 
+@pytest.mark.asyncio
+async def test_preferred_parent_forwards_to_list_changed_files() -> None:
+    """The in-path PR-review gate's cross-team parent (see ``diff``) must
+    reach ``list_changed_files`` so the validator never analyzes files
+    inherited from the wrong-team derived base."""
+    svc = _service()
+    _bind(svc, "_workspace_for_branch", AsyncMock(return_value=Path("/tmp/ws")))
+    changed = AsyncMock(return_value=[])
+    _bind(svc, "list_changed_files", changed)
+    actor_id = uuid4()
+    await svc.conventions_check_for_task(
+        actor_id,
+        _task("feature/frontend/root--cell"),
+        preferred_parent="feature/main_pm/root",
+    )
+    changed.assert_awaited_once_with(
+        branch_name="feature/frontend/root--cell",
+        actor_agent_id=actor_id,
+        preferred_parent="feature/main_pm/root",
+    )
+
+
 @pytest.mark.asyncio
 async def test_validator_timeout_fails_closed_and_reaps(
     tmp_path: Path, monkeypatch: pytest.MonkeyPatch

tests/unit/services/test_git_diff_base_fallback.py

@@ -169,7 +169,7 @@ async def test_diff_targets_origin_head_in_foreign_clone() -> None:
     # the fetches authenticate (unauth fails on private repos).
     svc._resolve_head_ref.assert_awaited_once_with(Path("/tmp/qa-ws"), _BR, token="tok")
     svc._resolve_diff_base.assert_awaited_once_with(
-        Path("/tmp/qa-ws"), _BR, token="tok"
+        Path("/tmp/qa-ws"), _BR, token="tok", preferred_parent=None
     )
 
 
@@ -192,7 +192,7 @@ async def test_list_changed_files_targets_origin_head_in_foreign_clone() -> None
     assert files == ["README.md", "src/app.py"]
     assert captured == [["diff", "--name-only", f"origin/master...origin/{_BR}"]]
     svc._resolve_diff_base.assert_awaited_once_with(
-        Path("/tmp/qa-ws"), _BR, token="tok"
+        Path("/tmp/qa-ws"), _BR, token="tok", preferred_parent=None
     )
 
 
@@ -217,6 +217,93 @@ async def test_diff_honours_explicit_base_with_resolved_head() -> None:
     svc._resolve_diff_base.assert_not_awaited()
 
 
+# ---------------------------------------------------------------------------
+# In-path PR-review gate cross-team fix: an explicit ``preferred_parent``
+# (resolve_parent_branch's real parent-task branch) overrides the derived
+# parent_branch_for, fetched + qualified exactly like the derived one — and
+# falls back to the same repo-default when it was never pushed. An explicit
+# literal ``base`` (e.g. HEAD~1 above) still wins outright and ignores it.
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_resolve_diff_base_uses_preferred_parent_when_pushed() -> None:
+    svc = _git_service()
+    svc._run_git = AsyncMock()
+    svc._ref_exists = AsyncMock(return_value=True)
+    ws = Path("/tmp/ws")
+
+    base = await svc._resolve_diff_base(
+        ws,
+        "feature/frontend/f7d0a61a--e56e6543--e2b50b06",
+        preferred_parent="feature/main_pm/f7d0a61a--e56e6543",
+    )
+    # NOT the same-team derivation (feature/frontend/f7d0a61a--e56e6543).
+    assert base == "origin/feature/main_pm/f7d0a61a--e56e6543"
+
+
+@pytest.mark.asyncio
+async def test_resolve_diff_base_preferred_parent_falls_back_when_absent() -> None:
+    """A preferred_parent that was never pushed (unassembled branchless
+    parent) still falls back to the repo default branch — never crashes."""
+    svc = _git_service()
+    svc._run_git = AsyncMock()
+    svc._ref_exists = AsyncMock(return_value=False)
+    svc._default_branch_ref = AsyncMock(return_value="origin/master")
+    ws = Path("/tmp/ws")
+
+    base = await svc._resolve_diff_base(
+        ws, "feature/main_pm/f7d0a61a--e56e6543", preferred_parent="master"
+    )
+    assert base == "origin/master"
+    svc._default_branch_ref.assert_awaited_once()
+
+
+@pytest.mark.asyncio
+async def test_diff_threads_preferred_parent_into_resolve_diff_base() -> None:
+    """diff()/list_changed_files() forward preferred_parent only when base is
+    omitted — the gate's evidence-build path (no explicit base)."""
+    svc = _git_service()
+    svc._workspace_for_branch = AsyncMock(return_value=Path("/tmp/ws"))
+    svc._resolve_head_ref = AsyncMock(return_value=_BR)
+    svc._token_for_branch = AsyncMock(return_value="tok")
+    svc._ref_exists = AsyncMock(return_value=True)
+    svc._run_git = AsyncMock(
+        return_value=type("R", (), {"returncode": 0, "stdout": "diff body"})()
+    )
+
+    out = await svc.diff(branch_name=_BR, preferred_parent="feature/main_pm/root")
+    assert out == "diff body"
+    svc._run_git.assert_any_call(
+        Path("/tmp/ws"),
+        ["diff", f"origin/feature/main_pm/root...{_BR}"],
+        check=False,
+    )
+
+
+@pytest.mark.asyncio
+async def test_explicit_base_ignores_preferred_parent() -> None:
+    """An explicit literal base wins outright — preferred_parent is only
+    consulted when base is omitted."""
+    svc = _git_service()
+    svc._workspace_for_branch = AsyncMock(return_value=Path("/tmp/ws"))
+    svc._resolve_head_ref = AsyncMock(return_value=_BR)
+    svc._token_for_branch = AsyncMock(return_value=None)
+    svc._resolve_diff_base = AsyncMock(return_value="SHOULD_NOT_BE_USED")
+    captured: list[list[str]] = []
+
+    async def fake_run(_ws: Any, args: list[str], **_kw: Any) -> Any:
+        captured.append(args)
+        return type("R", (), {"returncode": 0, "stdout": ""})()
+
+    with patch.object(svc, "_run_git", new=fake_run):
+        await svc.diff(
+            branch_name=_BR, base="HEAD~1", preferred_parent="feature/main_pm/root"
+        )
+    assert captured == [["diff", f"HEAD~1...{_BR}"]]
+    svc._resolve_diff_base.assert_not_awaited()
+
+
 # ---------------------------------------------------------------------------
 # Task #168: the diff base must be CURRENT. In an inspecting clone
 # origin/HEAD is set, so _default_branch_ref early-returns the ref NAME