feat(lifecycle): inherit advanced upstream base on work re-claims (#644)

* feat(lifecycle): inherit advanced upstream base on work re-claims

A re-claim reused a branch cut at an earlier claim, so upstream work
merged since (UX/UI landing on the root after the cell branch was cut)
never reached BE/FE branches — divergence and avoidable conflicts.
_finalize_claim now merges the advanced base into the pre-existing branch
via the dependency-lineage merge: already-ancestor is a no-op, a conflict
aborts at the cut point and leaves a transition note steering the agent
to sync_branch, a clean merge logs an audit trail; never fails the claim.

Double-gated: by role (developer/cell_pm/main_pm — QA/documenter/gate
claims review the branch as pushed and never move it) AND by pre-claim
status (pending/needs_revision only — a PM's i_will_plan re-claim of its
own awaiting_pm_review task must not move a branch that already passed
QA + the PR gate). Fresh cuts already branch from the live remote base.
Cell-PM prompt now orders reading the upstream design docs before
planning.

* fix(lifecycle): extract base-inheritance gate predicate for xenon budget

The four-condition inline gate pushed _finalize_claim to cyclomatic
rank C; the quality gate caps blocks at B. The decision moves to a pure
module-level predicate, byte-for-byte the same logic.

---------

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

agents/prompts/roles/cell_pm.md

@@ -75,7 +75,7 @@ When the briefing carries `company_goals`, let the charter guide how you scope a
 ## Workflow
 
 0. **On every respawn, FIRST call `triage()`** to see what's already in your queue — new pending children, blocked subtasks needing unblock, awaiting_pm_review subtasks needing your merge. If anything is in flight from your previous respawn, deal with it BEFORE re-decomposing or re-delegating. Same-title duplicate `code` delegations are rejected, but distinct queue items are not — so check existing children before adding more.
-1. `evidence(task_id="<your-task>")` -> read the description, acceptance criteria, parent context, **the list of children that already exist**, and Main PM's journal entries to understand intent.
+1. `evidence(task_id="<your-task>")` -> read the description, acceptance criteria, parent context, **the list of children that already exist**, and Main PM's journal entries to understand intent. **If your cell implements work another cell designed (BE/FE building on UX/UI output), also read the design docs BEFORE planning**: the repo's `docs/` tree and the UX cell's committed design assets on your base branch (read-only git: `git log`/`git diff` on the root branch shows what UX merged). Your subtask descriptions must reference those deliverables so devs build against the actual design, not a guess — your branch inherits the merged upstream base on every (re)claim, so the design files are already in your tree.
 2. **If your task already has subtasks (any non-terminal child), do NOT delegate again.** You are being respawned to coordinate, not to re-decompose. Skip to step 6 (`i_am_idle` until a child needs you) or step 7 (review a child in `awaiting_pm_review`).
 3. `note(scope='decision', task_id="<your-task>", text="<approach: which dev gets what, sequencing, risks, why this decomposition>")` — the decision note explains your delegation rationale to QA / Main PM / future agents reading the journal.
 4. `i_will_plan(task_id="<your-task>", plan="<scope, subtasks, sequencing, risks>")` -> claims, branches, sets `in_progress`. **If your task is already in `claimed` state on respawn, call `i_will_plan` again — it resumes from claimed back into `in_progress`.**

roboco/services/task.py

@@ -387,6 +387,35 @@ def _append_capped(existing: str | None, addition: str) -> str:
 _CEO_REJECT_ACTUAL_CAP = 300
 _CEO_REJECT_EVIDENCE_CAP = 2000
 
+# Roles whose claim WORKS the branch (and so should inherit an advanced
+# upstream base into it). QA / documenter / PR-gate claims review the branch
+# exactly as the dev pushed it and must never move it.
+_BASE_INHERIT_ROLES = frozenset({"developer", "cell_pm", "main_pm"})
+
+# Pre-claim statuses where inheriting is safe: work is (re)starting, nothing
+# downstream has reviewed the branch yet. A PM's i_will_plan re-claim of its
+# own AWAITING_PM_REVIEW task must NOT inherit — that branch already passed
+# QA + the PR gate, and a silent base merge would put unreviewed content
+# under the PM's merge decision.
+_BASE_INHERIT_STATUSES = frozenset({TaskStatus.PENDING, TaskStatus.NEEDS_REVISION})
+
+
+def _should_inherit_base(
+    original_branch_name: str | None,
+    project_id: Any,
+    agent_role: str | None,
+    original_status: Any,
+) -> bool:
+    """True iff this claim should merge the advanced upstream base in:
+    a pre-existing branch on a project task, claimed by a WORK role, from a
+    pre-review status. Kept out of ``_finalize_claim`` for complexity budget."""
+    return bool(
+        original_branch_name
+        and project_id
+        and agent_role in _BASE_INHERIT_ROLES
+        and original_status in _BASE_INHERIT_STATUSES
+    )
+
 
 def _ceo_reject_finding_texts(reason: str) -> tuple[str, str | None]:
     """Split a CEO rejection reason into a ledger Finding's (actual, evidence).
@@ -2595,6 +2624,88 @@ class TaskService(BaseService):
             files=result.get("files"),
         )
 
+    async def _inherit_upstream_base(self, task: TaskTable, agent_id: UUID) -> None:
+        """Merge the task's advanced upstream base into its existing branch.
+
+        Reuses the dependency-lineage merge: an already-ancestor base is a
+        cheap no-op, a conflict aborts (branch left at its cut point) and
+        leaves a transition note steering the agent to ``sync_branch``.
+        Best-effort — never fails the claim.
+        """
+        from roboco.services.git import get_git_service
+        from roboco.services.project import get_project_service
+
+        try:
+            project = await get_project_service(self.session).get(
+                UUID(str(task.project_id))
+            )
+            if not project:
+                return
+            base_branch = await self._resolve_parent_branch(task, project)
+            branch_name = str(task.branch_name)
+            if not base_branch or base_branch == branch_name:
+                return
+            git_service = get_git_service(self.session)
+            workspace = await git_service.get_workspace(project.slug, agent_id)
+            result = await git_service.merge_dependency_lineage(
+                workspace,
+                require_uuid(task.id),
+                branch_name,
+                base_branch,
+                project_slug=project.slug,
+            )
+            status = result.get("status")
+            if status == "merged":
+                # The one path that changes branch content — leave a trail.
+                self.log.info(
+                    "upstream base inherited into task branch",
+                    task_id=str(task.id),
+                    branch_name=branch_name,
+                    base_branch=base_branch,
+                )
+            elif status == "conflict":
+                self._note_base_inheritance_conflict(task, base_branch, result)
+                await self.session.flush()
+            elif status not in ("already_ancestor", "missing_ref"):
+                # missing_ref is quiet: a merged-and-deleted parent branch
+                # simply has nothing left to inherit.
+                self.log.warning(
+                    "upstream base inheritance incomplete",
+                    task_id=str(task.id),
+                    base_branch=base_branch,
+                    status=status,
+                )
+        except Exception:
+            self.log.warning(
+                "upstream base inheritance errored",
+                task_id=str(task.id),
+                exc_info=True,
+            )
+
+    def _note_base_inheritance_conflict(
+        self, task: TaskTable, base_branch: str, result: dict[str, Any]
+    ) -> None:
+        """Log + note a base-inheritance merge conflict for the assignee."""
+        files = ", ".join(result.get("files") or []) or "unknown files"
+        note = (
+            f"Upstream base {base_branch!r} has advanced with changes that "
+            f"conflict with this branch ({task.branch_name!r}) in: {files}. "
+            f"Merge {base_branch!r} in by hand (sync_branch) before submitting."
+        )
+        existing = markers.get_transition_note(task, "base_inheritance_conflict")
+        markers.set_transition_note(
+            task,
+            "base_inheritance_conflict",
+            f"{existing}\n{note}" if existing else note,
+        )
+        self.log.warning(
+            "upstream base inheritance conflict",
+            task_id=str(task.id),
+            branch_name=task.branch_name,
+            base_branch=base_branch,
+            files=result.get("files"),
+        )
+
     async def _distinct_projects_for_task(self, task: TaskTable) -> list[UUID]:
         """The distinct projects a coordination root's map spans — one
         ``feature/main_pm/{root}`` integration branch each.
@@ -3213,6 +3324,19 @@ class TaskService(BaseService):
             raise
         await self.session.refresh(task)
 
+        # Upstream base inheritance: a re-claim reuses a branch cut at an
+        # earlier claim, so upstream work merged since (e.g. UX/UI landing on
+        # the root after this cell branch was cut) never reaches it. Merge the
+        # advanced base back in before the agent spawns. Work-claims only —
+        # QA/doc/gate claims must review the branch as pushed, and a PM's
+        # i_will_plan re-claim of its own AWAITING_PM_REVIEW task must not
+        # move a branch that already passed QA + the PR gate. A fresh cut
+        # (original_branch_name unset) already branches from the live base.
+        if _should_inherit_base(
+            original_branch_name, task.project_id, agent_role, original_status
+        ):
+            await self._inherit_upstream_base(task, agent_id)
+
         await self._create_work_session_if_needed(task, agent_id, agent_role)
 
         bg_task = asyncio.create_task(self._inject_proactive_context(task, agent_id))

tests/unit/services/test_task_base_inheritance.py

@@ -0,0 +1,269 @@
+"""Upstream base inheritance on re-claim.
+
+A re-claim reuses a branch cut at an earlier claim; upstream work merged
+since (a sibling UX/UI cell landing on the root, master advancing under a
+root) never reached it, so BE/FE branches diverged from design work they
+were meant to build on. ``_finalize_claim`` now merges the advanced base
+into the pre-existing branch on WORK claims (developer / cell_pm / main_pm)
+via ``_inherit_upstream_base`` — QA/doc/gate claims never move the branch,
+and a fresh cut already branches from the live remote base.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+from unittest.mock import AsyncMock, MagicMock, patch
+from uuid import uuid4
+
+import pytest
+from roboco.foundation.policy.content import markers
+from roboco.models.base import TaskStatus
+from roboco.services.task import TaskService
+
+
+def _service() -> TaskService:
+    svc = TaskService.__new__(TaskService)
+    svc.log = MagicMock()
+    session = MagicMock()
+    session.flush = AsyncMock()
+    session.refresh = AsyncMock()
+    svc.session = session
+    return svc
+
+
+def _claim_task(
+    branch_name: str | None,
+    project_id: Any = None,
+    status: TaskStatus = TaskStatus.PENDING,
+) -> MagicMock:
+    return MagicMock(
+        id=uuid4(),
+        project_id=project_id if project_id is not None else uuid4(),
+        branch_name=branch_name,
+        status=status,
+        assigned_to=None,
+        claimed_by=None,
+        claimed_at=None,
+        last_heartbeat_at=None,
+        active_claimant_id=None,
+        orchestration_markers={},
+    )
+
+
+def _wire_finalize(svc: TaskService) -> AsyncMock:
+    """Stub every _finalize_claim collaborator; return the inherit mock."""
+    object.__setattr__(svc, "_set_original_developer_context", MagicMock())
+    object.__setattr__(svc, "_validate_and_set_status", MagicMock())
+    object.__setattr__(svc, "_emit_status_transition_audit", MagicMock())
+    object.__setattr__(svc, "_ensure_branch_for_task", AsyncMock(return_value="b"))
+    object.__setattr__(svc, "_create_work_session_if_needed", AsyncMock())
+    object.__setattr__(svc, "_inject_proactive_context", AsyncMock())
+    object.__setattr__(svc, "_CLAIMABLE_STATUSES", {TaskStatus.PENDING})
+    inherit = AsyncMock()
+    object.__setattr__(svc, "_inherit_upstream_base", inherit)
+    return inherit
+
+
+def _agent(role: str) -> MagicMock:
+    agent = MagicMock()
+    agent.role.value = role
+    return agent
+
+
+# ---------------------------------------------------------------------------
+# _finalize_claim gating
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_dev_reclaim_with_existing_branch_inherits() -> None:
+    svc = _service()
+    inherit = _wire_finalize(svc)
+    task = _claim_task("feature/backend/AAA--BBB")
+
+    await svc._finalize_claim(task, _agent("developer"), uuid4())
+
+    inherit.assert_awaited_once()
+
+
+@pytest.mark.asyncio
+async def test_pm_reclaim_with_existing_branch_inherits() -> None:
+    svc = _service()
+    inherit = _wire_finalize(svc)
+    task = _claim_task("feature/main_pm/AAA")
+
+    await svc._finalize_claim(task, _agent("cell_pm"), uuid4())
+
+    inherit.assert_awaited_once()
+
+
+@pytest.mark.asyncio
+async def test_pm_review_queue_reclaim_never_inherits() -> None:
+    """A PM's i_will_plan re-claim of its own AWAITING_PM_REVIEW task must
+    not move a branch that already passed QA + the PR gate — a silent base
+    merge there would put unreviewed content under the merge decision."""
+    svc = _service()
+    inherit = _wire_finalize(svc)
+    task = _claim_task("feature/main_pm/AAA", status=TaskStatus.AWAITING_PM_REVIEW)
+
+    await svc._finalize_claim(task, _agent("cell_pm"), uuid4())
+
+    inherit.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_needs_revision_reclaim_inherits() -> None:
+    """A bounced task re-claimed by its dev is the flagship inherit case."""
+    svc = _service()
+    inherit = _wire_finalize(svc)
+    task = _claim_task("feature/backend/AAA--BBB", status=TaskStatus.NEEDS_REVISION)
+
+    await svc._finalize_claim(task, _agent("developer"), uuid4())
+
+    inherit.assert_awaited_once()
+
+
+@pytest.mark.asyncio
+async def test_qa_claim_never_moves_the_branch() -> None:
+    svc = _service()
+    inherit = _wire_finalize(svc)
+    task = _claim_task("feature/backend/AAA--BBB")
+
+    await svc._finalize_claim(task, _agent("qa"), uuid4())
+
+    inherit.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_fresh_branch_skips_inheritance() -> None:
+    """No pre-claim branch → the fresh cut already builds on the live base."""
+    svc = _service()
+    inherit = _wire_finalize(svc)
+    task = _claim_task(None)
+
+    await svc._finalize_claim(task, _agent("developer"), uuid4())
+
+    inherit.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_branchless_coordination_skips_inheritance() -> None:
+    svc = _service()
+    inherit = _wire_finalize(svc)
+    task = _claim_task("feature/main_pm/AAA")
+    task.project_id = None
+
+    await svc._finalize_claim(task, _agent("main_pm"), uuid4())
+
+    inherit.assert_not_awaited()
+
+
+# ---------------------------------------------------------------------------
+# _inherit_upstream_base behavior
+# ---------------------------------------------------------------------------
+
+
+def _patched_deps(
+    svc: TaskService,
+    merge_status: dict[str, Any] | Exception,
+    parent_branch: str = "feature/main_pm/AAA",
+) -> tuple[Any, Any, AsyncMock]:
+    project = MagicMock(slug="roboco-api")
+    proj_svc = MagicMock()
+    proj_svc.get = AsyncMock(return_value=project)
+    git_svc = MagicMock()
+    git_svc.get_workspace = AsyncMock(return_value=MagicMock())
+    merge = AsyncMock(
+        side_effect=merge_status
+        if isinstance(merge_status, Exception)
+        else [merge_status]
+    )
+    git_svc.merge_dependency_lineage = merge
+    object.__setattr__(
+        svc, "_resolve_parent_branch", AsyncMock(return_value=parent_branch)
+    )
+    return proj_svc, git_svc, merge
+
+
+@pytest.mark.asyncio
+async def test_inherit_merges_parent_branch() -> None:
+    svc = _service()
+    task = _claim_task("feature/backend/AAA--BBB")
+    proj_svc, git_svc, merge = _patched_deps(svc, {"status": "merged"})
+
+    with (
+        patch(
+            "roboco.services.project.get_project_service",
+            MagicMock(return_value=proj_svc),
+        ),
+        patch("roboco.services.git.get_git_service", MagicMock(return_value=git_svc)),
+    ):
+        await svc._inherit_upstream_base(task, uuid4())
+
+    merge.assert_awaited_once()
+    args = merge.await_args
+    assert args is not None
+    assert args.args[2] == "feature/backend/AAA--BBB"
+    assert args.args[3] == "feature/main_pm/AAA"
+    assert task.orchestration_markers == {}, "clean merge leaves no note"
+
+
+@pytest.mark.asyncio
+async def test_inherit_conflict_notes_the_task() -> None:
+    svc = _service()
+    task = _claim_task("feature/backend/AAA--BBB")
+    proj_svc, git_svc, _ = _patched_deps(
+        svc, {"status": "conflict", "files": ["a.py", "b.py"]}
+    )
+
+    with (
+        patch(
+            "roboco.services.project.get_project_service",
+            MagicMock(return_value=proj_svc),
+        ),
+        patch("roboco.services.git.get_git_service", MagicMock(return_value=git_svc)),
+    ):
+        await svc._inherit_upstream_base(task, uuid4())
+
+    note = markers.get_transition_note(task, "base_inheritance_conflict")
+    assert note is not None
+    assert "a.py, b.py" in note
+    assert "sync_branch" in note
+
+
+@pytest.mark.asyncio
+async def test_inherit_skips_when_base_equals_branch() -> None:
+    svc = _service()
+    task = _claim_task("feature/main_pm/AAA")
+    proj_svc, git_svc, merge = _patched_deps(
+        svc, {"status": "merged"}, parent_branch="feature/main_pm/AAA"
+    )
+
+    with (
+        patch(
+            "roboco.services.project.get_project_service",
+            MagicMock(return_value=proj_svc),
+        ),
+        patch("roboco.services.git.get_git_service", MagicMock(return_value=git_svc)),
+    ):
+        await svc._inherit_upstream_base(task, uuid4())
+
+    merge.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_inherit_never_fails_the_claim() -> None:
+    svc = _service()
+    task = _claim_task("feature/backend/AAA--BBB")
+    proj_svc, git_svc, _ = _patched_deps(svc, RuntimeError("network down"))
+
+    with (
+        patch(
+            "roboco.services.project.get_project_service",
+            MagicMock(return_value=proj_svc),
+        ),
+        patch("roboco.services.git.get_git_service", MagicMock(return_value=git_svc)),
+    ):
+        await svc._inherit_upstream_base(task, uuid4())  # must not raise
+
+    svc.log.warning.assert_called()