feat(env-branches): per-project ordered environment ladder (replaces default_branch) (#534)

* [env-bran] EnvSyncEngine: orchestrator-side prod→dev cascade (default-off)

- EnvSyncEngine mirrors CiWatchEngine: cascade ladder_pairs top-down via
  GitHub merges API; clean→auto-push lower rung, conflict→one sync PR +
  tracked MAIN_PM task + stop. Never pushes prod (lower rung is never prod
  by construction).
- GitService.sync_env_branch (merges API) + open_sync_pr (idempotent) +
  _env_merge_status/_post_sync_pr helpers (constants for 201/204/409).
- TaskService.ENV_SYNC_SOURCE + list_open_env_sync_tasks (per-repo dedup).
- config env_sync_enabled/_interval_seconds(1800)/_max_open_tasks(3)/_max_per_cycle(1).
- Orchestrator 4-touch registration + _load_env_sync_set (ladder+token opt-in).
- Feature-flags card + settings FEATURE_FLAGS entry for ROBOCO_ENV_SYNC_ENABLED.

* [env-bran] Panel: environment ladder editor + types + validation

- EnvironmentRung type + environments on Project/ProjectCreate/ProjectUpdate.
- EnvironmentLadderEditor (plain useState, add/remove/up-down reorder, head/
  prod labels) reused by create + edit project dialogs.
- validateLadder (non-empty name+branch, no duplicate branches) shared,
  toast.error on submit; empty editor => null => inherits default_branch shim.
- default_branch input kept with override-hint; API client passthrough.
- 6 unit tests for validateLadder.

* [env-bran] Tests + gate green: env ladder, EnvSyncEngine, promotion chain

- tests/unit/models/test_env_branches.py: shim, head/prod, ladder_pairs,
  promotion_chain, normalize (20 tests)
- tests/integration/services/test_env_sync_engine.py: cascade clean/conflict/
  missing_ref/tokenless/degenerate/caps/dedup/disabled (9 tests, DB)
- tests/integration/test_migration_env_branches.py: 073 defaults null + round-trip
- tests/unit/services/test_release_executor*.py: add env_chain=[] to
  _ReleaseContext constructions (promotion_chain field is now required)
- tests/unit/runtime/test_orchestrator_shutdown_drain.py: register _env_sync_task
  in the stop()-drain fixture (new named background loop)
- roboco/services/git.py: revert _project_head_branch rename back to
  _project_default_branch (modify-in-place per plan); the rename in the
  consumers commit broke ~15 unit-test mocks that bind the original name
- roboco/services/env_sync_engine.py + models/env_branches.py: ruff format
- roboco/api/schemas/project.py: trailing-newline format

Backend gate green (13013 passed / 439 skipped), mypy clean, ruff clean.
Panel gate green (typecheck/lint/522 tests).

* [env-bran] fix: add env_chain to _ReleaseContext in e2e smoke (CI red)

The release-executor promotion_chain change made _ReleaseContext.env_chain
required. I fixed the three unit/release test files but missed the
construction in tests/e2e_smoke/test_background_engines.py:98 — my local
gate ran 'mypy roboco/' (excludes tests/) and I skipped 'make e2e-smoke',
so CI's mypy-on-tests + the e2e runtime job caught it instead of me.

Verified locally with the CI-equivalent gates:
  uv run mypy roboco/ tests/   -> 1170 files, clean
  ROBOCO_E2E_SMOKE=1 uv run pytest tests/e2e_smoke -> 50 passed, 1 skipped

* [env-bran] fix: extract _ensure_prod_fetched to clear xenon rank C (CI red)

_production_assess grew past xenon --max-absolute B (rank C) when the
env-branches prod-tip fetch added an if/try/except branch. Extracted the
fetch-with-fallback into _ensure_prod_fetched (degan+fetch paths), moved
_run_git to the module-level import. Local make quality green (all gates
incl xenon/vulture/deptry/import-linter/foundation-check).

---------

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

tests/e2e_smoke/test_background_engines.py

@@ -102,6 +102,7 @@ async def test_h24_wait_for_ci_polls_through_non_success(
         git_url="",
         git_prefix=[],
         ci_workflow=None,
+        env_chain=[],
     )
     ops = _GitReleaseOps(session=MagicMock(), ctx=ctx)
     sha = "abc123"

tests/integration/services/test_env_sync_engine.py

@@ -0,0 +1,269 @@
+"""EnvSyncEngine — cascade prod->head; conflict opens a PR + task, never prod.
+
+Mirrors the ci-watch engine test: seeds projects + agents, mocks the GitService
+(the merges API + PR open are GitHub calls) so the cascade is driven by the
+fake's queued statuses, and asserts the engine's contract:
+
+* clean cascade (merged / already_ancestor) opens nothing,
+* a conflict opens ONE sync PR + tracked task and stops the cascade,
+* a tokenless project is skipped,
+* per-cycle + rolling caps + per-repo dedup are honoured,
+* the cascade target is never prod (the lower rung of every pair is non-prod).
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Any
+from uuid import uuid4
+
+import pytest
+from roboco.config import settings
+from roboco.db.tables import AgentTable, ProjectTable
+from roboco.foundation import identity as _foundation
+from roboco.models.base import AgentRole, AgentStatus, TaskStatus, Team
+from roboco.services import env_sync_engine as env_sync_module
+from roboco.services.env_sync_engine import get_env_sync_engine
+from roboco.services.task import ENV_SYNC_SOURCE, get_task_service
+
+if TYPE_CHECKING:
+    from sqlalchemy.ext.asyncio import AsyncSession
+
+SYSTEM_UUID = _foundation.AGENTS["system"].uuid
+MAIN_PM_UUID = _foundation.AGENTS["main-pm"].uuid
+
+# A two-rung ladder: head=dev (PR target), prod=master (release target).
+_LADDER = [
+    {"name": "head", "branch": "dev"},
+    {"name": "prod", "branch": "master"},
+]
+
+
+class _FakeGit:
+    """Stand-in for GitService: drives the cascade from queued statuses."""
+
+    def __init__(self, statuses: list[str], *, pr_number: int = 42) -> None:
+        self._statuses = list(statuses)
+        self.sync_calls: list[tuple[str, str, str]] = []
+        self.pr_calls: list[tuple[str, str, str, str]] = []
+        self._pr_number = pr_number
+
+    async def sync_env_branch(
+        self, slug: str, target_branch: str, source_branch: str
+    ) -> dict[str, Any]:
+        self.sync_calls.append((slug, target_branch, source_branch))
+        status = self._statuses.pop(0) if self._statuses else "already_ancestor"
+        return {"status": status}
+
+    async def open_sync_pr(
+        self, slug: str, source_branch: str, target_branch: str, body: str
+    ) -> dict[str, Any] | None:
+        self.pr_calls.append((slug, source_branch, target_branch, body))
+        return {
+            "number": self._pr_number,
+            "url": f"https://github.com/x/{slug}/pull/{self._pr_number}",
+        }
+
+
+async def _get_or_create_agent(
+    db: AsyncSession, agent_id: object, role: AgentRole, slug: str
+) -> None:
+    if await db.get(AgentTable, agent_id) is None:
+        db.add(
+            AgentTable(
+                id=agent_id,
+                name=slug,
+                slug=f"{slug}-{uuid4().hex[:8]}",
+                role=role,
+                team=None,
+                status=AgentStatus.ACTIVE,
+                model_config={},
+                system_prompt="x",
+                capabilities=[],
+                permissions={},
+                metrics={},
+            )
+        )
+        await db.flush()
+
+
+async def _seed_project(
+    db: AsyncSession,
+    slug: str,
+    git_url: str,
+    *,
+    environments: list[dict[str, str]] | None = _LADDER,
+    token: str | None = "fake-encrypted-token",
+) -> ProjectTable:
+    project = ProjectTable(
+        id=uuid4(),
+        name=slug,
+        slug=slug,
+        git_url=git_url,
+        assigned_cell=Team.BACKEND,
+        created_by=SYSTEM_UUID,
+        environments=environments,
+        git_token_encrypted=token,
+    )
+    db.add(project)
+    await db.flush()
+    return project
+
+
+@pytest.fixture(autouse=True)
+async def _enabled(db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(settings, "env_sync_enabled", True)
+    monkeypatch.setattr(settings, "env_sync_max_per_cycle", 5)
+    monkeypatch.setattr(settings, "env_sync_max_open_tasks", 5)
+    await _get_or_create_agent(db_session, SYSTEM_UUID, AgentRole.SYSTEM, "system")
+    await _get_or_create_agent(db_session, MAIN_PM_UUID, AgentRole.MAIN_PM, "main-pm")
+
+
+def _patch_git(monkeypatch: pytest.MonkeyPatch, fake: _FakeGit) -> None:
+    monkeypatch.setattr(env_sync_module, "get_git_service", lambda _session: fake)
+
+
+@pytest.mark.asyncio
+async def test_clean_cascade_opens_nothing(
+    db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    proj = await _seed_project(db_session, "clean-a", "https://github.com/x/a.git")
+    fake = _FakeGit(["merged"])
+    _patch_git(monkeypatch, fake)
+    created = await get_env_sync_engine(db_session).run_cycle([proj])
+    assert created == []
+    # prod(master) merged into head(dev) — one cascade step for the 2-rung ladder.
+    assert fake.sync_calls == [("clean-a", "dev", "master")]
+    assert fake.pr_calls == []
+
+
+@pytest.mark.asyncio
+async def test_already_ancestor_is_clean(
+    db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    proj = await _seed_project(db_session, "anc-a", "https://github.com/x/anc.git")
+    fake = _FakeGit(["already_ancestor"])
+    _patch_git(monkeypatch, fake)
+    assert await get_env_sync_engine(db_session).run_cycle([proj]) == []
+    assert fake.sync_calls == [("anc-a", "dev", "master")]
+
+
+@pytest.mark.asyncio
+async def test_conflict_opens_pr_and_task_then_stops(
+    db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    # 4-rung ladder so a conflict on the FIRST (topmost) pair stops the cascade
+    # before reaching head — proving it does not cascade a dirty merge downward.
+    proj = await _seed_project(
+        db_session,
+        "conf-a",
+        "https://github.com/x/conf.git",
+        environments=[
+            {"name": "head", "branch": "dev"},
+            {"name": "qa", "branch": "qa"},
+            {"name": "stag", "branch": "stag"},
+            {"name": "prod", "branch": "master"},
+        ],
+    )
+    fake = _FakeGit(["conflict", "merged"])  # only the first is consumed
+    _patch_git(monkeypatch, fake)
+    created = await get_env_sync_engine(db_session).run_cycle([proj])
+
+    assert len(created) == 1
+    task = created[0]
+    assert task.source == ENV_SYNC_SOURCE
+    assert task.status == TaskStatus.PENDING
+    assert task.project_id == proj.id
+    # The cascade stopped at the first conflict: only one sync_env_branch call.
+    assert len(fake.sync_calls) == 1
+    # The sync PR targets the lower (non-prod) rung of the conflicted pair.
+    assert len(fake.pr_calls) == 1
+    _slug, _source_branch, target_branch, _body = fake.pr_calls[0]
+    assert target_branch != "master"  # never prod
+
+
+@pytest.mark.asyncio
+async def test_missing_ref_skips_without_pr(
+    db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    proj = await _seed_project(db_session, "miss-a", "https://github.com/x/miss.git")
+    fake = _FakeGit(["missing_ref"])
+    _patch_git(monkeypatch, fake)
+    created = await get_env_sync_engine(db_session).run_cycle([proj])
+    assert created == []
+    assert fake.pr_calls == []
+
+
+@pytest.mark.asyncio
+async def test_tokenless_project_skipped(
+    db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    proj = await _seed_project(
+        db_session, "notok", "https://github.com/x/notok.git", token=None
+    )
+    fake = _FakeGit(["merged"])
+    _patch_git(monkeypatch, fake)
+    created = await get_env_sync_engine(db_session).run_cycle([proj])
+    assert created == []
+    assert fake.sync_calls == []  # never attempted
+
+
+@pytest.mark.asyncio
+async def test_degenerate_ladder_skipped(
+    db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    # Single-rung ladder (head==prod) has no pairs to cascade.
+    proj = await _seed_project(
+        db_session,
+        "single",
+        "https://github.com/x/single.git",
+        environments=[{"name": "prod", "branch": "master"}],
+    )
+    fake = _FakeGit(["merged"])
+    _patch_git(monkeypatch, fake)
+    created = await get_env_sync_engine(db_session).run_cycle([proj])
+    assert created == []
+    assert fake.sync_calls == []
+
+
+@pytest.mark.asyncio
+async def test_per_cycle_cap(
+    db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    monkeypatch.setattr(settings, "env_sync_max_per_cycle", 1)
+    p1 = await _seed_project(db_session, "cap-1", "https://github.com/x/c1.git")
+    p2 = await _seed_project(db_session, "cap-2", "https://github.com/x/c2.git")
+    fake = _FakeGit(["conflict", "conflict"])
+    _patch_git(monkeypatch, fake)
+    created = await get_env_sync_engine(db_session).run_cycle([p1, p2])
+    assert len(created) == 1  # capped at one per cycle
+
+
+@pytest.mark.asyncio
+async def test_deduped_per_repo(
+    db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """A repo with an open env_sync task is skipped until the PR resolves."""
+    proj = await _seed_project(db_session, "dedup", "https://github.com/x/d.git")
+    fake = _FakeGit(["conflict"])
+    _patch_git(monkeypatch, fake)
+    first = await get_env_sync_engine(db_session).run_cycle([proj])
+    assert len(first) == 1
+    # Second cycle, same repo still has the open task -> deduped (no new PR).
+    fake2 = _FakeGit(["conflict"])
+    _patch_git(monkeypatch, fake2)
+    second = await get_env_sync_engine(db_session).run_cycle([proj])
+    assert second == []
+    assert fake2.sync_calls == []  # cascade paused at the conflicted rung
+    assert len(await get_task_service(db_session).list_open_env_sync_tasks()) == 1
+
+
+@pytest.mark.asyncio
+async def test_disabled_is_noop(
+    db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    monkeypatch.setattr(settings, "env_sync_enabled", False)
+    proj = await _seed_project(db_session, "off", "https://github.com/x/off.git")
+    fake = _FakeGit(["merged"])
+    _patch_git(monkeypatch, fake)
+    assert await get_env_sync_engine(db_session).run_cycle([proj]) == []
+    assert fake.sync_calls == []

tests/integration/test_migration_env_branches.py

@@ -0,0 +1,74 @@
+"""Per-project environment ladder column (migration 073).
+
+Migration 073 adds ``projects.environments`` (JSONB null). The real
+upgrade/downgrade chain is verified separately against a throwaway Postgres;
+these assertions guard the resulting schema shape and a value round-trip,
+mirroring ``test_migration_dep_update.py``.
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+from uuid import uuid4
+
+import pytest
+from roboco.db.tables import AgentTable, ProjectTable
+from roboco.models import AgentRole, AgentStatus, Team
+from sqlalchemy import select
+
+if TYPE_CHECKING:
+    from sqlalchemy.ext.asyncio import AsyncSession
+
+
+async def _seed_project(db_session: AsyncSession) -> ProjectTable:
+    agent = AgentTable(
+        id=uuid4(),
+        name="Dev",
+        slug=f"be-dev-{uuid4().hex[:8]}",
+        role=AgentRole.DEVELOPER,
+        team=Team.BACKEND,
+        status=AgentStatus.ACTIVE,
+        model_config={},
+        system_prompt="dev",
+        capabilities=[],
+        permissions={},
+        metrics={},
+    )
+    db_session.add(agent)
+    await db_session.flush()
+    project = ProjectTable(
+        id=uuid4(),
+        name="L-Proj",
+        slug=f"l-proj-{uuid4().hex[:8]}",
+        git_url="https://example.com/r.git",
+        assigned_cell=Team.BACKEND,
+        created_by=agent.id,
+    )
+    db_session.add(project)
+    await db_session.flush()
+    return project
+
+
+@pytest.mark.asyncio
+async def test_environments_defaults_null(db_session: AsyncSession) -> None:
+    project = await _seed_project(db_session)
+    assert project.environments is None
+
+
+@pytest.mark.asyncio
+async def test_environments_round_trip(db_session: AsyncSession) -> None:
+    project = await _seed_project(db_session)
+    project.environments = [
+        {"name": "head", "branch": "dev"},
+        {"name": "prod", "branch": "master"},
+    ]
+    await db_session.flush()
+    row = (
+        await db_session.execute(
+            select(ProjectTable).where(ProjectTable.id == project.id)
+        )
+    ).scalar_one()
+    assert row.environments == [
+        {"name": "head", "branch": "dev"},
+        {"name": "prod", "branch": "master"},
+    ]

tests/unit/models/test_env_branches.py

@@ -0,0 +1,230 @@
+"""env_branches — shim, ladder pairs, head/prod resolution, normalization.
+
+Pure domain helpers (pydantic-only); no DB. The read-time shim synthesizes a
+degenerate single-branch ladder from ``default_branch`` when ``environments``
+is null, so every consumer behaves identically until a real ladder is declared.
+"""
+
+from __future__ import annotations
+
+import pytest
+from roboco.models.env_branches import (
+    EnvRung,
+    effective_environments,
+    head_branch,
+    ladder_pairs,
+    normalize_environments,
+    prod_branch,
+    promotion_chain,
+)
+
+
+class _Proj:
+    """Duck-typed project row (matches Project + ProjectTable surface)."""
+
+    def __init__(
+        self,
+        *,
+        default_branch: str = "master",
+        environments: list[dict[str, str]] | None = None,
+    ) -> None:
+        self.default_branch = default_branch
+        self.environments = environments
+
+
+# --- effective_environments shim ------------------------------------------
+
+
+def test_null_environments_synthesizes_degenerate_ladder() -> None:
+    proj = _Proj(default_branch="slave")
+    rungs = effective_environments(proj)
+    assert [(r.name, r.branch) for r in rungs] == [("head", "slave"), ("prod", "slave")]
+
+
+def test_empty_environments_synthesizes_degenerate_ladder() -> None:
+    proj = _Proj(default_branch="master", environments=[])
+    rungs = effective_environments(proj)
+    assert [(r.name, r.branch) for r in rungs] == [
+        ("head", "master"),
+        ("prod", "master"),
+    ]
+
+
+def test_missing_default_branch_falls_back_to_master() -> None:
+    proj = _Proj(default_branch="")  # falsy default_branch
+    assert head_branch(proj) == "master"
+    assert prod_branch(proj) == "master"
+
+
+def test_set_environments_returned_as_is_preserving_order() -> None:
+    proj = _Proj(
+        environments=[
+            {"name": "head", "branch": "dev"},
+            {"name": "qa", "branch": "qa"},
+            {"name": "prod", "branch": "master"},
+        ]
+    )
+    rungs = effective_environments(proj)
+    assert [r.branch for r in rungs] == ["dev", "qa", "master"]
+    assert all(isinstance(r, EnvRung) for r in rungs)
+
+
+# --- head_branch / prod_branch --------------------------------------------
+
+
+def test_head_and_prod_single_rung() -> None:
+    proj = _Proj(environments=[{"name": "prod", "branch": "master"}])
+    assert head_branch(proj) == "master"
+    assert prod_branch(proj) == "master"
+
+
+def test_head_and_prod_two_rungs() -> None:
+    proj = _Proj(
+        environments=[
+            {"name": "head", "branch": "slave"},
+            {"name": "prod", "branch": "master"},
+        ]
+    )
+    assert head_branch(proj) == "slave"
+    assert prod_branch(proj) == "master"
+
+
+def test_head_and_prod_four_rungs() -> None:
+    proj = _Proj(
+        environments=[
+            {"name": "head", "branch": "dev"},
+            {"name": "qa", "branch": "qa"},
+            {"name": "stag", "branch": "stag"},
+            {"name": "prod", "branch": "master"},
+        ]
+    )
+    assert head_branch(proj) == "dev"
+    assert prod_branch(proj) == "master"
+
+
+# --- ladder_pairs (prod -> head cascade) -----------------------------------
+
+
+def test_ladder_pairs_empty_for_single_rung() -> None:
+    assert (
+        ladder_pairs(_Proj(environments=[{"name": "prod", "branch": "master"}])) == []
+    )
+
+
+def test_ladder_pairs_two_rungs() -> None:
+    proj = _Proj(
+        environments=[
+            {"name": "head", "branch": "dev"},
+            {"name": "prod", "branch": "master"},
+        ]
+    )
+    pairs = ladder_pairs(proj)
+    assert [(u.branch, lower.branch) for u, lower in pairs] == [("master", "dev")]
+
+
+def test_ladder_pairs_four_rungs_top_down() -> None:
+    proj = _Proj(
+        environments=[
+            {"name": "head", "branch": "dev"},
+            {"name": "qa", "branch": "qa"},
+            {"name": "stag", "branch": "stag"},
+            {"name": "prod", "branch": "master"},
+        ]
+    )
+    # [(prod, stag), (stag, qa), (qa, head)] — merge upper into lower.
+    pairs = ladder_pairs(proj)
+    assert [(u.branch, lower.branch) for u, lower in pairs] == [
+        ("master", "stag"),
+        ("stag", "qa"),
+        ("qa", "dev"),
+    ]
+
+
+def test_ladder_pairs_lower_rung_is_never_prod() -> None:
+    """The cascade's target is never prod by construction — only CEO merges prod."""
+    proj = _Proj(
+        environments=[
+            {"name": "head", "branch": "dev"},
+            {"name": "qa", "branch": "qa"},
+            {"name": "prod", "branch": "master"},
+        ]
+    )
+    prod_name = prod_branch(proj)
+    for _upper, lower in ladder_pairs(proj):
+        assert lower.branch != prod_name
+
+
+# --- promotion_chain (full-chain release promotion) ----------------------
+
+
+def test_promotion_chain_empty_for_degenerate_ladder() -> None:
+    """head==prod => nothing to promote (no-op release promotion)."""
+    assert promotion_chain(_Proj(default_branch="master")) == []
+    assert (
+        promotion_chain(_Proj(environments=[{"name": "prod", "branch": "master"}]))
+        == []
+    )
+
+
+def test_promotion_chain_two_rungs() -> None:
+    proj = _Proj(
+        environments=[
+            {"name": "head", "branch": "dev"},
+            {"name": "prod", "branch": "master"},
+        ]
+    )
+    assert promotion_chain(proj) == ["dev"]
+
+
+def test_promotion_chain_four_rungs_head_first_excluding_prod() -> None:
+    proj = _Proj(
+        environments=[
+            {"name": "head", "branch": "dev"},
+            {"name": "qa", "branch": "qa"},
+            {"name": "stag", "branch": "stag"},
+            {"name": "prod", "branch": "master"},
+        ]
+    )
+    assert promotion_chain(proj) == ["dev", "qa", "stag"]
+
+
+# --- normalize_environments ------------------------------------------------
+
+
+def test_normalize_none_returns_none() -> None:
+    assert normalize_environments(None) is None
+    assert normalize_environments([]) is None
+
+
+def test_normalize_strips_and_preserves_order() -> None:
+    out = normalize_environments(
+        [{"name": " head ", "branch": " dev "}, {"name": "prod", "branch": "master"}]
+    )
+    assert out == [
+        {"name": "head", "branch": "dev"},
+        {"name": "prod", "branch": "master"},
+    ]
+
+
+def test_normalize_rejects_empty_name() -> None:
+    with pytest.raises(ValueError, match="non-empty name"):
+        normalize_environments([{"name": "", "branch": "dev"}])
+
+
+def test_normalize_rejects_empty_branch() -> None:
+    with pytest.raises(ValueError, match="non-empty name"):
+        normalize_environments(
+            [{"name": "head", "branch": "  "}]
+        )  # branch trimmed to empty
+
+
+def test_normalize_rejects_duplicate_branch() -> None:
+    with pytest.raises(ValueError, match="duplicate environment branch"):
+        normalize_environments(
+            [{"name": "head", "branch": "dev"}, {"name": "prod", "branch": "dev"}]
+        )
+
+
+def test_normalize_accepts_envrung_models() -> None:
+    out = normalize_environments([EnvRung(name="head", branch="dev")])
+    assert out == [{"name": "head", "branch": "dev"}]

tests/unit/runtime/test_orchestrator_shutdown_drain.py

@@ -50,6 +50,7 @@ def _make_orchestrator() -> AgentOrchestrator:
         "_self_heal_task",
         "_ci_watch_task",
         "_dep_update_task",
+        "_env_sync_task",
         "_release_manager_task",
         "_x_mentions_task",
         "_roadmap_engine_task",

tests/unit/services/test_release_executor.py

@@ -70,6 +70,9 @@ class _FakeOps:
         # instance (not via __init__ — keeps the constructor under the arg-count
         # gate) by tests that exercise the retry path.
         self._existing_sha: str | None = None
+        # env-chain promotion failure message; set on the instance (same arg-
+        # count-gate reason) by the promotion-failure test.
+        self._promote_raises: str | None = None
         self.calls: list[str] = []
         self.bumped_plan: list[str] | None = None
         self.bumped_version: str | None = None
@@ -79,6 +82,11 @@ class _FakeOps:
         self.calls.append("check")
         return self._already
 
+    async def promote_env_chain(self) -> None:
+        self.calls.append("promote")
+        if self._promote_raises is not None:
+            raise RuntimeError(self._promote_raises)
+
     async def release_commit_sha(self, _version: str) -> str | None:
         # Half-landed detection: a prior `chore(release): {version}` commit
         # already on the branch means a publish_failed retry must NOT re-run the
@@ -127,6 +135,7 @@ async def test_green_path_publishes_once() -> None:
     assert ops.calls.count("publish") == _ONE
     assert ops.calls == [
         "check",
+        "promote",
         "bump",
         "changelog",
         "gate",
@@ -204,6 +213,22 @@ async def test_publish_failure_returns_structured_publish_failed() -> None:
     assert ops.calls.count("publish") == _ONE
 
 
+@pytest.mark.asyncio
+async def test_promotion_failure_aborts_before_bump() -> None:
+    """A RuntimeError from promote_env_chain (a merge conflict in the
+    head->...->prod chain) becomes a structured ``promotion_failed`` result —
+    fail-closed: the bump/changelog/gate/commit/publish pipeline never runs."""
+    ops = _FakeOps()
+    ops._promote_raises = "env-chain promotion failed: non-fast-forward"
+    result = await ReleaseExecutor(ops).execute(_report())
+    assert result.status == "promotion_failed"
+    assert result.commit_sha is None
+    assert result.release_url is None
+    assert "bump" not in ops.calls
+    assert "commit" not in ops.calls
+    assert "publish" not in ops.calls
+
+
 def test_release_result_carries_outcome_fields() -> None:
     result = ReleaseResult(
         status="published",
@@ -291,6 +316,7 @@ async def test_wait_for_ci_scoped_to_release_commit_not_branch_latest(
         git_url="x",
         git_prefix=[],
         ci_workflow="ci.yml",
+        env_chain=[],
     )
     ops = _GitReleaseOps(session=MagicMock(), ctx=ctx)
     ok = await ops.wait_for_ci(commit_sha)
@@ -341,6 +367,7 @@ async def test_wait_for_ci_polls_through_rerun(
         git_url="x",
         git_prefix=[],
         ci_workflow="ci.yml",
+        env_chain=[],
     )
     ops = _GitReleaseOps(session=MagicMock(), ctx=ctx)
     ok = await ops.wait_for_ci(commit_sha)
@@ -387,6 +414,7 @@ async def test_wait_for_ci_exhausts_window_on_persistent_failure(
         git_url="x",
         git_prefix=[],
         ci_workflow="ci.yml",
+        env_chain=[],
     )
     ops = _GitReleaseOps(session=MagicMock(), ctx=ctx)
     ok = await ops.wait_for_ci(commit_sha)
@@ -515,6 +543,7 @@ async def test_release_push_argv_uses_extraheader_not_url_token(
         git_url=git_url,
         git_prefix=git_prefix,
         ci_workflow=None,
+        env_chain=[],
     )
     ops = _GitReleaseOps(session=MagicMock(), ctx=ctx)
     sha = await ops.commit_and_push("0.13.0")

tests/unit/services/test_release_executor_commit_fail_closed.py

@@ -21,6 +21,7 @@ def _ctx() -> _ReleaseContext:
         git_url="https://github.com/o/roboco",
         git_prefix=[],
         ci_workflow=None,
+        env_chain=[],
     )