[feature] sync_branch dev verb — gate-level branch rebase (Phase B1)

Raw shell git is denied to agents (Bash(git:*) base deny), so a developer
whose branch fell behind its base had no gate-level rebase — only the
CEO/PM-only /rebase HTTP route. sync_branch is the dev verb that wraps the
rebase through the gate (traced + evidenced), so the 'everything goes through
the gates' invariant holds.

- lifecycle: IntentSpec sync_branch (dev-only, ownership-gated, composes=(),
  git-only — no DB transition); _next_hint_synced helper.
- GitService.sync_task_branch: rebase task.branch_name onto its resolved base
  via rebase_onto_base (fetch + rebase + force-with-lease push).
- Choreographer.sync_branch + _sync_branch_preflight_rejection: not_found /
  unknown-role / spec-gate / no-branch / protected-base guards, then the git
  op; conflicts abort (no force-push) and steer to resolve-by-hand; git failure
  steers to i_am_blocked.
- HTTP route /api/v1/flow/developer/sync_branch + SyncBranchRequest schema.
- MCP tool sync_branch(task_id) + _TOOLS registration (manifest auto-propagates
  via intents_for_role(Role.DEVELOPER)).

Tests: intent spec (5), choreographer handler (8: happy/conflicts/not_found/
not_authorized/no-branch/protected-base/git-failure/audit), route (1), MCP (1).
ruff + mypy roboco/ tests/ clean; unit suite green (DB-fixture errors env-only).

roboco/api/routes/v1/flow_dev.py

@@ -15,6 +15,7 @@ from roboco.api.schemas.v1.flow import (
     IWillWorkOnRequest,
     OpenPrRequest,
     ResumeRequest,
+    SyncBranchRequest,
     UnclaimRequest,
 )
 from roboco.services.gateway.choreographer import Choreographer
@@ -121,6 +122,17 @@ async def resume(
     return envelope_to_response(env, request)
 
 
+@router.post("/sync_branch")
+async def sync_branch(
+    request: Request,
+    body: SyncBranchRequest,
+    x_agent_id: _AgentIdHeader,
+    choreographer: _ChoreographerDep,
+) -> dict:
+    env = await choreographer.sync_branch(x_agent_id, body.task_id)
+    return envelope_to_response(env, request)
+
+
 @router.post("/i_am_idle")
 async def i_am_idle(
     request: Request,

roboco/api/schemas/v1/flow.py

@@ -114,6 +114,17 @@ class ResumeRequest(BaseModel):
     task_id: UUID
 
 
+class SyncBranchRequest(BaseModel):
+    """HTTP body for the dev `sync_branch` verb.
+
+    Rebases the task's branch onto its resolved base (parent branch) through the
+    gate, so a developer whose branch has fallen behind can re-sync without raw
+    git (which is denied to agents). Git-only — no DB state transition.
+    """
+
+    task_id: UUID
+
+
 class IAmIdleRequest(BaseModel):
     """Empty request body."""
 

roboco/foundation/policy/lifecycle.py

@@ -740,6 +740,13 @@ def _next_hint_open_pr(_t: Any) -> str:
     return "PR opened; call i_am_done(task_id, notes='...') when self-verified"
 
 
+def _next_hint_synced(_t: Any) -> str:
+    return (
+        "branch synced onto its base; continue editing + commit(message),"
+        " then open_pr(task_id) / i_am_done(task_id)"
+    )
+
+
 def _next_hint_after_claim(_t: Any) -> str:
     return (
         "edit + commit(message) for each meaningful change,"
@@ -1008,6 +1015,23 @@ _INTENT_VERBS: dict[str, IntentSpec] = {
         side_effects=(),
         next_hint=_next_hint_idle,
     ),
+    "sync_branch": IntentSpec(
+        name="sync_branch",
+        allowed_roles=_DEV_ROLES,
+        description=(
+            "Rebase your task's branch onto its current base THROUGH the gate"
+            " (raw git is denied). Use when your branch has fallen behind its"
+            " base — e.g. a sibling task's PR merged into the parent branch"
+            " while you worked. Fetches origin, rebases head onto base, and"
+            " force-pushes (with-lease). No DB state change. On conflicts the"
+            " rebase is aborted and the conflicted files are returned — resolve"
+            " by hand, commit, then sync_branch again."
+        ),
+        composes=(),  # git-only verb — no DB transition; the handler runs the git op
+        extra_preconditions=(PRECONDITION_OWNERSHIP,),
+        side_effects=(),
+        next_hint=_next_hint_synced,
+    ),
     "i_am_blocked": IntentSpec(
         name="i_am_blocked",
         allowed_roles=frozenset(_DEV_ROLES | _QA_ROLES | _DOC_ROLES),

roboco/mcp/flow_server.py

@@ -333,6 +333,21 @@ def resume(task_id: str) -> dict[str, Any]:
     return _post(_role_path("resume"), {"task_id": task_id})
 
 
+def sync_branch(task_id: str) -> dict[str, Any]:
+    """Re-sync your branch onto its base through the gate.
+
+    Rebases the task's branch onto its resolved parent/base branch (fetch +
+    rebase + force-with-lease push). Use this when your branch has fallen
+    behind its base and you need to pick up merged work before continuing —
+    raw git is denied, so this is the gate-level way to rebase. No lifecycle
+    transition: after it returns, keep editing + commit, then open_pr /
+    i_am_done as normal. On ``conflicts`` status the envelope's ``next`` tells
+    you the rebase aborted and your branch is unchanged — resolve the conflict
+    in your working tree first (the gate does not force a conflicted rebase).
+    """
+    return _post(_role_path("sync_branch"), {"task_id": task_id})
+
+
 def i_am_idle() -> dict[str, Any]:
     """Report no more work. Soft-blocks if you have unread A2A/mentions."""
     return _post(_role_path("i_am_idle"), {})
@@ -602,6 +617,7 @@ _TOOLS: dict[str, Any] = {
     "unclaim": unclaim,
     "reassign": reassign,
     "resume": resume,
+    "sync_branch": sync_branch,
     "i_am_idle": i_am_idle,
     # qa — keys are the public MCP tool names (what agents see and prompts
     # advertise). `pass`/`fail` are Python keywords so the IntentSpec uses

roboco/services/gateway/choreographer/_impl.py

@@ -3310,6 +3310,156 @@ class Choreographer:
             context_briefing=briefing,
         ).with_introspection(task=after, role=role_str)
 
+    async def sync_branch(self, agent_id: UUID, task_id: UUID) -> Envelope:
+        """Rebase the caller's task branch onto its current base THROUGH the gate.
+
+        Raw shell git is denied to agents (the ``Bash(git:*)`` base deny), so a
+        developer whose branch fell behind its base — a sibling's PR merged
+        into the parent branch while they worked — had no gate-level rebase,
+        only the CEO/PM-only ``/rebase`` HTTP route. ``sync_branch`` is that
+        gate verb: it resolves the task's base via
+        ``merge_chain.resolve_parent_branch``, guards against rebasing into a
+        protected branch, then rebases + force-pushes (with-lease) via
+        ``GitService.sync_task_branch``. Git-only (no DB state change), like
+        ``open_pr``; the spec gate checks role + ownership, then the handler
+        guards branch + base, then runs the git op. Conflicts abort the rebase
+        (no force-push) and return the conflicted files — resolve by hand,
+        commit, then sync_branch again.
+        """
+        t = await self.task.get(task_id)
+        briefing = await self._briefing_for(agent_id, task_id, task=t)
+        agent = await self.task.agent_for(agent_id)
+        role_str = str(agent.role) if agent is not None else "developer"
+        # Preflight: not_found / unknown-role / spec-gate / no-branch /
+        # protected-base. Returns the rejection (None when all clear) AND the
+        # resolved base_branch the git op needs (empty when rejected).
+        rejection, base_branch = await self._sync_branch_preflight_rejection(
+            agent_id, task_id, t, agent, role_str, briefing
+        )
+        if rejection is not None:
+            return await self._emit_rejection(
+                rejection, agent_id=agent_id, task_id=task_id, verb="sync_branch"
+            )
+        try:
+            result = await self.git.sync_task_branch(
+                t, base_branch=base_branch, actor_agent_id=agent_id
+            )
+        except Exception as exc:
+            return await self._emit_rejection(
+                Envelope.invalid_state(
+                    message=f"sync_branch failed: {exc}",
+                    remediate=(
+                        "the git rebase could not complete; escalate via"
+                        " i_am_blocked(reason='...') with the error"
+                    ),
+                    context_briefing=briefing,
+                ).with_introspection(task=t, role=role_str),
+                agent_id=agent_id,
+                task_id=task_id,
+                verb="sync_branch",
+            )
+        # Heartbeat — the agent is actively working the task.
+        await self._touch(task_id)
+        status = str(result.get("status", "unknown"))
+        evidence = {
+            "rebase": result,
+            "base_branch": base_branch,
+            "head_branch": str(t.branch_name),
+        }
+        if status == "conflicts":
+            # The rebase was aborted (no force-push); tell the dev to resolve.
+            next_hint = (
+                f"sync_branch hit conflicts on {result.get('files', [])};"
+                " resolve by hand, commit(message='...'), then sync_branch again"
+            )
+        else:
+            next_hint = spec_module._INTENT_VERBS["sync_branch"].next_hint(t)
+        return Envelope.ok(
+            status=str(t.status),
+            task_id=str(task_id),
+            next=next_hint,
+            evidence=evidence,
+            context_briefing=briefing,
+        ).with_introspection(task=t, role=role_str)
+
+    async def _sync_branch_preflight_rejection(
+        self,
+        agent_id: UUID,
+        task_id: UUID,
+        t: Any,
+        agent: Any,
+        role_str: str,
+        briefing: dict[str, Any],
+    ) -> tuple[Envelope | None, str]:
+        """Role + spec-gate + branch/base guards for ``sync_branch``.
+
+        Returns ``(rejection_envelope, base_branch)``. When all guards pass the
+        rejection is ``None`` and ``base_branch`` is the resolved merge target
+        the handler hands to ``GitService.sync_task_branch``; on any guard
+        failure the envelope is set and ``base_branch`` is empty. Extracted so
+        ``sync_branch`` stays under the PLR0911 return budget (mirrors
+        ``_open_pr_preflight_rejection``).
+        """
+        if t is None:
+            return (
+                Envelope.not_found(message=f"task {task_id} not found"),
+                "",
+            )
+        try:
+            role = spec_module.Role(role_str)
+        except ValueError:
+            return (
+                Envelope.not_authorized(
+                    message=f"unknown role '{role_str}'",
+                    remediate="role is not declared in the lifecycle spec",
+                    context_briefing=briefing,
+                ).with_introspection(task=t, role=role_str),
+                "",
+            )
+        spec_ctx = spec_module.Context(
+            actor_id=agent_id,
+            actor_slug=getattr(agent, "slug", None) if agent is not None else None,
+        )
+        decision = spec_module.can_invoke_intent(role, "sync_branch", t, spec_ctx)
+        if not decision.allowed:
+            return (
+                Envelope.from_decision(decision, briefing=briefing).with_introspection(
+                    task=t, role=role_str
+                ),
+                "",
+            )
+        # The task must carry a branch (claimed/in_progress) — there is nothing
+        # to sync before the branch is cut, and branchless coordination roots
+        # carry none.
+        if not t.branch_name:
+            return (
+                Envelope.invalid_state(
+                    message=f"task {task_id} has no branch_name to sync",
+                    remediate="call i_will_work_on(task_id) first to cut a branch",
+                    context_briefing=briefing,
+                ).with_introspection(task=t, role=role_str),
+                "",
+            )
+        base_branch = await resolve_parent_branch(t, self.task)
+        # Defense-in-depth: agents never rebase into a protected/default branch
+        # or a ``-``-prefixed (shell-injection) ref. A dev task's base is its
+        # parent (cell-task) branch, so this should never fire — but never let a
+        # rebase reach master through a branchless-parent fallback.
+        if base_branch.startswith("-") or base_branch in ("master", "main"):
+            return (
+                Envelope.invalid_state(
+                    message=f"resolved base branch '{base_branch}' is protected",
+                    remediate=(
+                        "the task's base resolved to master/main; sync_branch"
+                        " refuses to rebase into a protected branch — escalate"
+                        " via i_am_blocked(reason='...') if your base is wrong"
+                    ),
+                    context_briefing=briefing,
+                ).with_introspection(task=t, role=role_str),
+                "",
+            )
+        return None, base_branch
+
     async def i_am_idle(self, agent_id: UUID) -> Envelope:
         """Report no more work. Soft-block if there are unread A2As or @mentions.
 

roboco/services/git.py

@@ -3695,6 +3695,43 @@ class GitService(BaseService):
             git_token=git_token,
         )
 
+    async def sync_task_branch(
+        self,
+        task: Any,
+        *,
+        base_branch: str,
+        actor_agent_id: UUID | None = None,
+    ) -> dict[str, Any]:
+        """Rebase a task's branch onto ``base_branch`` through the gate.
+
+        Task-keyed twin of :meth:`rebase_pr_for_task`: ``head_branch`` is the
+        task's own ``branch_name`` and ``base_branch`` is supplied by the caller
+        (the choreographer resolves it via ``merge_chain.resolve_parent_branch``),
+        so this works BEFORE a PR exists — a developer mid-work whose branch
+        fell behind its base (a sibling's PR merged into the parent branch) can
+        rebase through the dev ``sync_branch`` verb instead of the CEO/PM-only
+        ``/rebase`` HTTP route. Mirrors ``rebase_pr_for_task``'s workspace/token
+        resolution and delegates to :meth:`rebase_onto_base`, returning the same
+        classification dict (``rebased`` / ``superseded`` / ``conflicts``).
+
+        The caller MUST ensure ``base_branch`` is not a protected branch —
+        agents never rebase into master/main; the choreographer guards this.
+        """
+        if not task.branch_name:
+            raise ValueError("sync_task_branch requires a task with a branch_name")
+        project = await self._project_for_task(task)
+        if project is None:
+            raise NotFoundError("Project for task", str(task.id))
+        workspace_agent_id = self._resolve_workspace_agent_id(task, actor_agent_id)
+        workspace = await self.get_workspace(project.slug, agent_id=workspace_agent_id)
+        git_token = await self._get_project_token_or_raise(project.slug)
+        return await self.rebase_onto_base(
+            workspace,
+            head_branch=task.branch_name,
+            base_branch=base_branch,
+            git_token=git_token,
+        )
+
     async def close_pull_request(
         self,
         pr_number: int,

tests/unit/api/routes/v1/test_flow_dev.py

@@ -205,3 +205,22 @@ async def test_resume_dispatches_task_id() -> None:
     )
     assert resp.status_code == _HTTP_200
     mock_chore.resume.assert_awaited_once()
+
+
+@pytest.mark.asyncio
+async def test_sync_branch_dispatches_task_id() -> None:
+    """POST sync_branch forwards task_id to Choreographer.sync_branch (git-only)."""
+    mock_chore = MagicMock()
+    mock_chore.sync_branch = AsyncMock(
+        return_value=_make_envelope(status="ok", task_id=_TASK_ID)
+    )
+    client = TestClient(_build_app(mock_chore))
+    resp = client.post(
+        "/api/v1/flow/developer/sync_branch",
+        json={"task_id": _TASK_ID},
+        headers=_HEADERS,
+    )
+    assert resp.status_code == _HTTP_200
+    mock_chore.sync_branch.assert_awaited_once()
+    # the only positional arg beyond x_agent_id is task_id
+    assert str(mock_chore.sync_branch.call_args.args[1]) == _TASK_ID

tests/unit/foundation/policy/test_sync_branch_intent.py

@@ -0,0 +1,77 @@
+"""IntentSpec for the dev `sync_branch` verb (multi-level sequencing Phase B1).
+
+Raw shell git is denied to agents by design (the `Bash(git:*)` base deny), so a
+developer whose branch has fallen behind its base had no gate-level way to
+rebase — only the CEO/PM-only `/rebase` HTTP route. `sync_branch` is the dev
+verb that wraps the git rebase through the gate (traced + evidenced), so the
+"everything goes through the gates" invariant holds. These tests lock the spec
+declaration: dev-only, ownership-gated, composes nothing (git-only, no DB
+transition), and present in the dev flow.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from uuid import uuid4
+
+from roboco.foundation.identity import Role
+from roboco.foundation.policy.lifecycle import (
+    Context,
+    Decision,
+    can_invoke_intent,
+    intents_for_role,
+)
+from roboco.services.gateway.role_config import _DEV_FLOW
+
+
+def test_sync_branch_is_a_dev_flow_verb() -> None:
+    # Declared with _DEV_ROLES, so intents_for_role propagates it into the dev
+    # flow automatically — no role_config edit needed (the spec is canon).
+    assert "sync_branch" in intents_for_role(Role.DEVELOPER)
+    assert "sync_branch" in _DEV_FLOW
+
+
+def test_sync_branch_is_dev_only() -> None:
+    # QA / documenter / PM cannot call it — it's a developer's branch-sync verb.
+    for role in (Role.QA, Role.DOCUMENTER, Role.CELL_PM, Role.MAIN_PM):
+        assert "sync_branch" not in intents_for_role(role), (
+            f"{role} must not get sync_branch"
+        )
+
+
+@dataclass
+class _Task:
+    assigned_to: object = None
+
+
+def test_sync_branch_requires_ownership() -> None:
+    # PRECONDITION_OWNERSHIP (rejection_kind='not_authorized') gates it: a task
+    # assigned to another agent rejects as not_authorized, not a tracing gap.
+    owner = uuid4()
+    other = uuid4()
+    task = _Task(assigned_to=other)
+    decision = can_invoke_intent(
+        Role.DEVELOPER, "sync_branch", task, Context(actor_id=owner)
+    )
+    assert not decision.allowed
+    assert decision.rejection_kind == "not_authorized"
+
+
+def test_sync_branch_allowed_when_owner() -> None:
+    # composes=() and the only precondition is ownership, so the owner passes the
+    # spec gate. (The choreographer handler does the git work + branch/base
+    # guards separately.)
+    owner = uuid4()
+    task = _Task(assigned_to=owner)
+    decision = can_invoke_intent(
+        Role.DEVELOPER, "sync_branch", task, Context(actor_id=owner)
+    )
+    assert isinstance(decision, Decision)
+    assert decision.allowed
+
+
+def test_sync_branch_unknown_to_other_role_rejects() -> None:
+    # A role not in allowed_roles is rejected as not_authorized (role gating).
+    decision = can_invoke_intent(Role.QA, "sync_branch", _Task(), Context())
+    assert not decision.allowed
+    assert decision.rejection_kind == "not_authorized"

tests/unit/gateway/test_sync_branch.py

@@ -0,0 +1,247 @@
+"""sync_branch rebases the caller's task branch onto its base THROUGH the gate.
+
+Multi-level sequencing Phase B1. Raw shell git is denied to agents
+(``Bash(git:*)`` base deny), so a developer whose branch fell behind its base
+had no gate-level rebase — only the CEO/PM-only ``/rebase`` HTTP route.
+``sync_branch`` is the dev verb that wraps the rebase through the gate
+(traced + evidenced), so the "everything goes through the gates" invariant
+holds. These tests pin the handler:
+
+- happy path: git.sync_task_branch runs, evidence carries the rebase result,
+  heartbeat fires
+- conflicts: rebase aborted, next_hint points the dev at resolve-by-hand
+- not_found: unknown task id
+- not_authorized: only the current claimant can sync (ownership gate)
+- no branch: branchless / not-yet-claimed task → invalid_state, steer to
+  i_will_work_on
+- protected base: resolved base == master/main → invalid_state (defense-in-depth)
+- git failure: sync_task_branch raises → invalid_state, steer to i_am_blocked
+"""
+
+from __future__ import annotations
+
+from unittest.mock import AsyncMock, MagicMock, patch
+from uuid import uuid4
+
+import pytest
+from roboco.services.gateway.choreographer import Choreographer, ChoreographerDeps
+
+
+def _make_deps(**overrides: object) -> ChoreographerDeps:
+    base: dict[str, object] = {
+        "task": AsyncMock(),
+        "work_session": AsyncMock(),
+        "git": AsyncMock(),
+        "a2a": AsyncMock(),
+        "journal": AsyncMock(),
+        "audit": AsyncMock(),
+        "evidence_repo": AsyncMock(),
+    }
+    base.update(overrides)
+    repo = base["evidence_repo"]
+    assert isinstance(repo, AsyncMock)
+    for method in (
+        "list_unread_a2a",
+        "list_unread_mentions",
+        "list_pending_notifications",
+        "task_metadata_gaps",
+        "recent_team_activity",
+        "blockers_in_lane",
+    ):
+        getattr(repo, method).return_value = []
+    return ChoreographerDeps(**base)
+
+
+_BRANCH = "feature/backend/abc12345"
+_BASE = "feature/backend/parent12345"
+
+
+def _task(*, tid: object, aid: object, branch: str | None = _BRANCH) -> MagicMock:
+    return MagicMock(
+        id=tid,
+        status="in_progress",
+        assigned_to=aid,
+        branch_name=branch,
+    )
+
+
+@pytest.mark.asyncio
+async def test_sync_branch_rebases_and_returns_evidence() -> None:
+    aid = uuid4()
+    tid = uuid4()
+    t = _task(tid=tid, aid=aid)
+    task_svc = AsyncMock()
+    task_svc.get.return_value = t
+    task_svc.agent_for.return_value = MagicMock(role="developer", team="backend")
+    git_svc = AsyncMock()
+    git_svc.sync_task_branch.return_value = {
+        "status": "rebased",
+        "commits_rebased": 3,
+    }
+    deps = _make_deps(task=task_svc, git=git_svc)
+    c = Choreographer(deps)
+
+    with patch(
+        "roboco.services.gateway.choreographer._impl.resolve_parent_branch",
+        new=AsyncMock(return_value=_BASE),
+    ):
+        env = await c.sync_branch(aid, tid)
+
+    git_svc.sync_task_branch.assert_awaited_once_with(
+        t, base_branch=_BASE, actor_agent_id=aid
+    )
+    assert env.error is None
+    assert env.evidence is not None
+    assert env.evidence["base_branch"] == _BASE
+    assert env.evidence["head_branch"] == _BRANCH
+    assert env.evidence["rebase"]["status"] == "rebased"
+    task_svc.heartbeat.assert_awaited_once_with(tid)
+
+
+@pytest.mark.asyncio
+async def test_sync_branch_conflicts_aborts_and_steers_to_resolve() -> None:
+    aid = uuid4()
+    tid = uuid4()
+    t = _task(tid=tid, aid=aid)
+    task_svc = AsyncMock()
+    task_svc.get.return_value = t
+    task_svc.agent_for.return_value = MagicMock(role="developer", team="backend")
+    git_svc = AsyncMock()
+    git_svc.sync_task_branch.return_value = {
+        "status": "conflicts",
+        "files": ["src/a.py", "src/b.py"],
+    }
+    deps = _make_deps(task=task_svc, git=git_svc)
+    c = Choreographer(deps)
+
+    with patch(
+        "roboco.services.gateway.choreographer._impl.resolve_parent_branch",
+        new=AsyncMock(return_value=_BASE),
+    ):
+        env = await c.sync_branch(aid, tid)
+
+    assert env.error is None
+    assert env.next is not None
+    assert "resolve by hand" in env.next
+    assert "sync_branch again" in env.next
+
+
+@pytest.mark.asyncio
+async def test_sync_branch_not_found_for_unknown_task() -> None:
+    aid = uuid4()
+    tid = uuid4()
+    task_svc = AsyncMock()
+    task_svc.get.return_value = None
+    deps = _make_deps(task=task_svc)
+    c = Choreographer(deps)
+
+    env = await c.sync_branch(aid, tid)
+
+    assert env.error == "not_found"
+
+
+@pytest.mark.asyncio
+async def test_sync_branch_rejects_when_not_claimant() -> None:
+    aid = uuid4()
+    other = uuid4()
+    tid = uuid4()
+    t = _task(tid=tid, aid=other)
+    task_svc = AsyncMock()
+    task_svc.get.return_value = t
+    task_svc.agent_for.return_value = MagicMock(role="developer", team="backend")
+    git_svc = AsyncMock()
+    deps = _make_deps(task=task_svc, git=git_svc)
+    c = Choreographer(deps)
+
+    env = await c.sync_branch(aid, tid)
+
+    # PRECONDITION_OWNERSHIP rejects a non-owner as not_authorized.
+    assert env.error == "not_authorized"
+    git_svc.sync_task_branch.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_sync_branch_no_branch_steers_to_i_will_work_on() -> None:
+    aid = uuid4()
+    tid = uuid4()
+    # branch_name=None — task not yet claimed / branchless coordination root.
+    t = _task(tid=tid, aid=aid, branch=None)
+    task_svc = AsyncMock()
+    task_svc.get.return_value = t
+    task_svc.agent_for.return_value = MagicMock(role="developer", team="backend")
+    git_svc = AsyncMock()
+    deps = _make_deps(task=task_svc, git=git_svc)
+    c = Choreographer(deps)
+
+    env = await c.sync_branch(aid, tid)
+
+    assert env.error == "invalid_state"
+    assert "i_will_work_on" in (env.remediate or "")
+    git_svc.sync_task_branch.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_sync_branch_refuses_protected_base() -> None:
+    aid = uuid4()
+    tid = uuid4()
+    t = _task(tid=tid, aid=aid)
+    task_svc = AsyncMock()
+    task_svc.get.return_value = t
+    task_svc.agent_for.return_value = MagicMock(role="developer", team="backend")
+    git_svc = AsyncMock()
+    deps = _make_deps(task=task_svc, git=git_svc)
+    c = Choreographer(deps)
+
+    # Defense-in-depth: a base that resolved to master must never be rebased into.
+    with patch(
+        "roboco.services.gateway.choreographer._impl.resolve_parent_branch",
+        new=AsyncMock(return_value="master"),
+    ):
+        env = await c.sync_branch(aid, tid)
+
+    assert env.error == "invalid_state"
+    assert "protected" in (env.message or "")
+    git_svc.sync_task_branch.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_sync_branch_git_failure_steers_to_i_am_blocked() -> None:
+    aid = uuid4()
+    tid = uuid4()
+    t = _task(tid=tid, aid=aid)
+    task_svc = AsyncMock()
+    task_svc.get.return_value = t
+    task_svc.agent_for.return_value = MagicMock(role="developer", team="backend")
+    git_svc = AsyncMock()
+    git_svc.sync_task_branch.side_effect = RuntimeError("network down")
+    deps = _make_deps(task=task_svc, git=git_svc)
+    c = Choreographer(deps)
+
+    with patch(
+        "roboco.services.gateway.choreographer._impl.resolve_parent_branch",
+        new=AsyncMock(return_value=_BASE),
+    ):
+        env = await c.sync_branch(aid, tid)
+
+    assert env.error == "invalid_state"
+    assert "i_am_blocked" in (env.remediate or "")
+
+
+@pytest.mark.asyncio
+async def test_sync_branch_rejection_writes_audit_row() -> None:
+    """Every rejection envelope must call audit.log_event (Task 6 contract)."""
+    aid = uuid4()
+    tid = uuid4()
+    task_svc = AsyncMock()
+    task_svc.get.return_value = None
+    audit_svc = AsyncMock()
+    deps = _make_deps(task=task_svc, audit=audit_svc)
+    c = Choreographer(deps)
+
+    env = await c.sync_branch(aid, tid)
+
+    assert env.error == "not_found"
+    audit_svc.log_event.assert_awaited_once()
+    kwargs = audit_svc.log_event.await_args.kwargs
+    assert kwargs["event_type"] == "gateway.rejected"
+    assert kwargs["details"]["verb"] == "sync_branch"

tests/unit/mcp_servers/test_flow_server.py

@@ -29,6 +29,7 @@ _FULL_MANIFEST = {
         "i_am_blocked",
         "unclaim",
         "resume",
+        "sync_branch",
         "i_am_idle",
         "claim_review",
         "pass",
@@ -214,6 +215,19 @@ def test_i_am_done_notes_defaults_to_empty(flow_module: types.ModuleType) -> Non
     assert kwargs["json"]["notes"] == ""
 
 
+def test_sync_branch_posts_to_dev_path(flow_module: types.ModuleType) -> None:
+    """sync_branch forwards task_id to /api/v1/flow/developer/sync_branch."""
+    fake_client = _make_fake_client({"status": "ok"})
+
+    with patch("httpx.Client", return_value=fake_client):
+        result = flow_module.sync_branch("task-abc")
+
+    assert result == {"status": "ok"}
+    args, kwargs = fake_client.post.call_args
+    assert "/api/v1/flow/developer/sync_branch" in args[0]
+    assert kwargs["json"] == {"task_id": "task-abc"}
+
+
 def test_i_am_blocked_sends_reason(flow_module: types.ModuleType) -> None:
     fake_client = _make_fake_client({"status": "blocked"})