+ Fallback head+prod branch when no environment ladder is set below.
+
+ {/* Environment ladder */}
+
diff --git a/panel/src/components/projects/environment-ladder-editor.tsx b/panel/src/components/projects/environment-ladder-editor.tsx
new file mode 100644
index 00000000..dbfe40ee
--- /dev/null
+++ b/panel/src/components/projects/environment-ladder-editor.tsx
@@ -0,0 +1,155 @@
+"use client";
+
+import { Button } from "@/components/ui/button";
+import { Input } from "@/components/ui/input";
+import { Label } from "@/components/ui/label";
+import {
+ Tooltip,
+ TooltipContent,
+ TooltipTrigger,
+} from "@/components/ui/tooltip";
+import { ArrowDown, ArrowUp, Plus, X } from "lucide-react";
+import type { EnvironmentRung } from "@/types";
+
+interface EnvironmentLadderEditorProps {
+ // null/empty => degenerate 1-rung ladder synthesized from default_branch.
+ rungs: EnvironmentRung[] | null;
+ onChange: (rungs: EnvironmentRung[] | null) => void;
+}
+
+// An empty editor (no rungs) means "inherit default_branch" via the backend
+// shim, so we keep a plain array internally and emit null when it empties.
+function toRungs(value: EnvironmentRung[] | null): EnvironmentRung[] {
+ return value ?? [];
+}
+
+export function EnvironmentLadderEditor({
+ rungs,
+ onChange,
+}: EnvironmentLadderEditorProps) {
+ const items = toRungs(rungs);
+
+ const emit = (next: EnvironmentRung[]) => {
+ onChange(next.length ? next : null);
+ };
+
+ const handleAdd = () => {
+ emit([...items, { name: "", branch: "" }]);
+ };
+
+ const handleRemove = (index: number) => {
+ emit(items.filter((_, i) => i !== index));
+ };
+
+ const handleUpdate = (index: number, field: keyof EnvironmentRung, value: string) => {
+ const next = items.map((r, i) => (i === index ? { ...r, [field]: value } : r));
+ emit(next);
+ };
+
+ const handleMove = (index: number, direction: -1 | 1) => {
+ const target = index + direction;
+ if (target < 0 || target >= items.length) return;
+ const next = [...items];
+ [next[index], next[target]] = [next[target], next[index]];
+ emit(next);
+ };
+
+ return (
+
+
+
+
+ {items.length} rung{items.length !== 1 ? "s" : ""}
+
+
+
+ {items.length > 0 && (
+
+ {items.map((rung, index) => {
+ const isFirst = index === 0;
+ const isLast = index === items.length - 1;
+ return (
+
+
+
+
+
+
+ Move up (toward head)
+
+
+
+
+
+ Move down (toward prod)
+
+
+
+ {isFirst ? "head" : isLast ? "prod" : `rung ${index + 1}`}
+
+
handleUpdate(index, "name", e.target.value)}
+ placeholder="Name (e.g. dev, qa, stag)"
+ className="flex-1 h-8"
+ />
+
handleUpdate(index, "branch", e.target.value)}
+ placeholder="Branch (e.g. dev, master)"
+ className="flex-1 h-8"
+ />
+
+
+
+
+ Remove this rung
+
+
+ );
+ })}
+
+ )}
+
+
+
+
+ Ordered top→bottom: the first rung is head (where dev PRs
+ land) and the last is prod (the release target). Leave
+ empty to inherit default branch for both — e.g.{" "}
+ dev → qa → stag → prod, or just prod for a
+ single-branch project. When set, this overrides default branch{" "}
+ for the PR target and the release target.
+
+
+ );
+}
\ No newline at end of file
diff --git a/panel/src/components/projects/ladder-validation.ts b/panel/src/components/projects/ladder-validation.ts
new file mode 100644
index 00000000..5ce5cf11
--- /dev/null
+++ b/panel/src/components/projects/ladder-validation.ts
@@ -0,0 +1,19 @@
+import type { EnvironmentRung } from "@/types";
+
+// Validate a ladder before submit. Returns an error string or null when valid.
+// null/empty is valid (inherits default_branch via the backend shim).
+export function validateLadder(rungs: EnvironmentRung[] | null): string | null {
+ if (!rungs || rungs.length === 0) return null;
+ const branches: string[] = [];
+ for (const rung of rungs) {
+ if (!rung.name.trim() || !rung.branch.trim()) {
+ return "Every environment rung needs both a name and a branch.";
+ }
+ const branch = rung.branch.trim();
+ if (branches.includes(branch)) {
+ return `Duplicate branch "${branch}" — each rung must target a unique branch.`;
+ }
+ branches.push(branch);
+ }
+ return null;
+}
\ No newline at end of file
diff --git a/panel/src/components/settings/feature-flags-card.tsx b/panel/src/components/settings/feature-flags-card.tsx
index 8dfc8932..eca39930 100644
--- a/panel/src/components/settings/feature-flags-card.tsx
+++ b/panel/src/components/settings/feature-flags-card.tsx
@@ -63,6 +63,8 @@ const FLAG_DESCRIPTIONS: Record
= {
"Watch every opted-in project's CI and open a fix task when its default branch goes red (per-project opt-in; never auto-merges).",
dep_update_enabled:
"Periodically probe opted-in projects for dependency updates and open an update task when a lockfile would change (per-project opt-in; never auto-merges).",
+ env_sync_enabled:
+ "Cascade each project's declared environment ladder prod→dev via GitHub's merges API so dev never falls behind prod; a conflicted rung opens a sync PR for you to merge (per-project opt-in; never pushes prod).",
release_manager_enabled:
"Run the deterministic release-readiness sweep and propose a release for you to approve or reject — it never publishes without your approval, and the executor is fail-closed on a red gate.",
org_memory_enabled:
diff --git a/panel/src/lib/api/projects.ts b/panel/src/lib/api/projects.ts
index 5172a47d..32ca5102 100644
--- a/panel/src/lib/api/projects.ts
+++ b/panel/src/lib/api/projects.ts
@@ -125,6 +125,7 @@ export const projectsApi = {
slug: project.slug,
git_url: project.git_url,
default_branch: project.default_branch ?? "main",
+ environments: project.environments ?? null,
protected_branches: project.protected_branches ?? ["main", "master"],
assigned_cell: project.assigned_cell,
has_git_token: !!project.git_token, // Mock token status
diff --git a/panel/src/types/index.ts b/panel/src/types/index.ts
index 3ee5c132..abd94cc7 100644
--- a/panel/src/types/index.ts
+++ b/panel/src/types/index.ts
@@ -1022,12 +1022,21 @@ export enum WorkSessionStatus {
ABANDONED = "abandoned",
}
+// One rung of a project's ordered environment ladder.
+export interface EnvironmentRung {
+ name: string;
+ branch: string;
+}
+
export interface Project {
id: string;
name: string;
slug: string;
git_url: string;
default_branch: string;
+ // Ordered environment ladder (first=head/PR-target, last=prod/release-target).
+ // Null/empty => degenerate 1-rung ladder synthesized from default_branch.
+ environments: EnvironmentRung[] | null;
protected_branches: string[];
assigned_cell: Team;
// Git authentication (token never exposed, only boolean indicator)
@@ -1063,6 +1072,8 @@ export interface ProjectCreate {
slug: string;
git_url: string;
default_branch?: string;
+ // Ordered environment ladder; null/empty inherits default_branch (shim).
+ environments?: EnvironmentRung[] | null;
protected_branches?: string[];
assigned_cell: Team;
// Git authentication (stored encrypted, never returned)
@@ -1079,6 +1090,8 @@ export interface ProjectUpdate {
name?: string;
git_url?: string;
default_branch?: string;
+ // Ordered environment ladder; null clears (reverts to default_branch shim).
+ environments?: EnvironmentRung[] | null;
protected_branches?: string[];
assigned_cell?: Team;
// Git authentication (empty string clears, undefined leaves unchanged)
diff --git a/roboco/api/schemas/project.py b/roboco/api/schemas/project.py
index ef01fa9a..91eaa916 100644
--- a/roboco/api/schemas/project.py
+++ b/roboco/api/schemas/project.py
@@ -295,4 +295,3 @@ def project_to_summary(
ci_watch_enabled=bool(project.ci_watch_enabled),
task_counts=task_counts,
)
-
diff --git a/roboco/config.py b/roboco/config.py
index 5f10e2c5..964f66a0 100644
--- a/roboco/config.py
+++ b/roboco/config.py
@@ -818,6 +818,37 @@ class Settings(BaseSettings):
description="Max dep_update tasks the loop may originate in one cycle.",
)
+ # Env-sync engine — cascades each opted-in project's env ladder prod→…→head
+ # (a clean merge auto-pushes to the lower rung; a conflict opens ONE sync PR)
+ # so dev never falls behind prod. Default-off; never pushes to prod (the
+ # cascade's lower/target rung is never prod by construction).
+ env_sync_enabled: bool = Field(
+ default=False,
+ description=(
+ "Master switch for the env-sync cascade loop. OFF by default; when "
+ "off the loop does not run. Only projects with a declared env "
+ "ladder (environments set) AND a git token participate."
+ ),
+ )
+ env_sync_interval_seconds: int = Field(
+ default=1800,
+ ge=60,
+ description="Seconds between env-sync cascade passes.",
+ )
+ env_sync_max_open_tasks: int = Field(
+ default=3,
+ ge=1,
+ description=(
+ "Rolling cap on concurrently-open env_sync conflict tasks across "
+ "all repos; the loop originates nothing more while this many are open."
+ ),
+ )
+ env_sync_max_per_cycle: int = Field(
+ default=1,
+ ge=1,
+ description="Max projects the env-sync loop may cascade in one cycle.",
+ )
+
# Gated release manager — at a logical point (accumulated unreleased changes
# past a threshold + green gate) the Secretary runs a deterministic readiness
# sweep and PROPOSES a release for the CEO to approve/reject. Default-off;
diff --git a/roboco/models/env_branches.py b/roboco/models/env_branches.py
index 883e138b..4852d218 100644
--- a/roboco/models/env_branches.py
+++ b/roboco/models/env_branches.py
@@ -87,6 +87,19 @@ def ladder_pairs(project: object) -> list[tuple[EnvRung, EnvRung]]:
return [(rungs[i], rungs[i - 1]) for i in range(len(rungs) - 1, 0, -1)]
+def promotion_chain(project: object) -> list[str]:
+ """Branches to merge into the prod checkout head->...->just-below-prod on a
+ CEO-gated release (the full-chain promotion).
+
+ Every rung except the last (prod), excluding any rung sharing the prod
+ branch — so a degenerate (head==prod) ladder yields ``[]`` (no-op). Order
+ is head-first: ``[dev, qa, stag]`` for ``[dev, qa, stag, master]``.
+ """
+ rungs = effective_environments(project)
+ prod = prod_branch(project)
+ return [r.branch for r in rungs[:-1] if r.branch != prod]
+
+
def normalize_environments(
value: list[dict[str, str]] | list[EnvRung] | None,
) -> list[dict[str, str]] | None:
diff --git a/roboco/runtime/orchestrator.py b/roboco/runtime/orchestrator.py
index a475faab..67f64ea1 100644
--- a/roboco/runtime/orchestrator.py
+++ b/roboco/runtime/orchestrator.py
@@ -1109,6 +1109,7 @@ class AgentOrchestrator:
self._self_heal_task: asyncio.Task | None = None
self._ci_watch_task: asyncio.Task | None = None
self._dep_update_task: asyncio.Task | None = None
+ self._env_sync_task: asyncio.Task | None = None
self._release_manager_task: asyncio.Task | None = None
self._x_mentions_task: asyncio.Task | None = None
self._roadmap_engine_task: asyncio.Task | None = None
@@ -1205,6 +1206,7 @@ class AgentOrchestrator:
self._self_heal_task = asyncio.create_task(self._self_heal_loop())
self._ci_watch_task = asyncio.create_task(self._ci_watch_loop())
self._dep_update_task = asyncio.create_task(self._dep_update_loop())
+ self._env_sync_task = asyncio.create_task(self._env_sync_loop())
self._release_manager_task = asyncio.create_task(self._release_manager_loop())
self._x_mentions_task = asyncio.create_task(self._x_mentions_poll_loop())
self._roadmap_engine_task = asyncio.create_task(self._roadmap_engine_loop())
@@ -1321,6 +1323,7 @@ class AgentOrchestrator:
self._self_heal_task,
self._ci_watch_task,
self._dep_update_task,
+ self._env_sync_task,
self._release_manager_task,
self._x_mentions_task,
self._roadmap_engine_task,
@@ -8137,6 +8140,72 @@ Start by:
await get_dep_update_engine(db).run_cycle(projects)
await db.commit()
+ async def _env_sync_loop(self) -> None:
+ """Env-sync: cascade prod→…→head so dev never falls behind prod.
+
+ Dormant by default — returns immediately unless ``env_sync_enabled``,
+ so a standard deployment adds zero behaviour. Each interval it loads the
+ opted-in projects (a declared env ladder + a git token), cascades the
+ ladder top-down via GitHub's merges API, and opens ONE sync PR + tracked
+ task per conflicted repo. Never pushes prod (the cascade's target is a
+ non-prod rung by construction), so "only the CEO merges master" holds.
+ """
+ if not settings.env_sync_enabled:
+ return
+ interval = settings.env_sync_interval_seconds
+ self._record_loop_heartbeat("env_sync", interval)
+ while self._running:
+ try:
+ await asyncio.sleep(interval)
+ await self._run_env_sync_cycle()
+ self._record_loop_heartbeat("env_sync", interval)
+ except asyncio.CancelledError:
+ break
+ except Exception:
+ logger.exception("env-sync cycle failed")
+
+ async def _run_env_sync_cycle(self) -> None:
+ """One env-sync pass: load the opted-in set, run the engine, commit.
+
+ Extracted from the loop so it is testable without the sleep. Warns when
+ env-sync is armed but no project has a declared env ladder — so a
+ misconfiguration isn't mistaken for "everything is in sync".
+ """
+ from roboco.db import get_db_context
+ from roboco.services.env_sync_engine import get_env_sync_engine
+
+ async with get_db_context() as db:
+ projects = await self._load_env_sync_set(db)
+ if not projects:
+ logger.warning(
+ "env-sync enabled but no project has a declared environment "
+ "ladder + git token — nothing to cascade"
+ )
+ return
+ await get_env_sync_engine(db).run_cycle(projects)
+ await db.commit()
+
+ async def _load_env_sync_set(self, db: Any) -> list[Any]:
+ """Projects opted into env-sync: a declared env ladder (more than one
+ rung, so there are pairs to cascade) + a git_url, one per repo.
+
+ A degenerate ladder (head==prod, one rung) has no pairs to cascade, so it
+ is excluded here — there is nothing to sync. Collapse to one canonical
+ project per repo: the engine's per-``git_url`` open-task dedup means a
+ monorepo's several cell-projects share one cascade anyway.
+ """
+ from roboco.models.env_branches import ladder_pairs
+ from roboco.services.project import get_project_service
+
+ projects = await get_project_service(db).list_all(active_only=True)
+ eligible = [
+ p for p in projects if getattr(p, "git_url", None) and ladder_pairs(p)
+ ]
+ return self._projects_one_per_key(
+ eligible,
+ key_fn=lambda p: (self._repo_key(str(getattr(p, "git_url", "") or "")),),
+ )
+
async def _release_manager_loop(self) -> None:
"""Gated release manager: at a logical point, propose a CEO-gated release.
diff --git a/roboco/services/env_sync_engine.py b/roboco/services/env_sync_engine.py
new file mode 100644
index 00000000..24dff6f9
--- /dev/null
+++ b/roboco/services/env_sync_engine.py
@@ -0,0 +1,202 @@
+"""Env-sync engine — cascade prod→…→head so dev never falls behind prod.
+
+Dormant by default (``env_sync_enabled``). For each opted-in project (a declared
+env ladder + a git token) it cascades the upper rungs into the lower ones top-down
+via GitHub's merges API: a clean merge auto-pushes to the lower rung, and a
+conflict opens ONE sync PR (upper→lower) and stops that project's cascade for the
+cycle. Like the other background engines it is conservative:
+
+* **Default OFF** (``env_sync_enabled``) — the orchestrator loop never starts.
+* **Never pushes prod** — the cascade's lower/target rung is never prod by
+ construction (``ladder_pairs`` yields ``(upper, lower)`` with lower ∈ the
+ non-prod rungs), so "only the CEO merges master" holds.
+* **Bounded + deduped per repo** — at most one open env_sync task per repo (a
+ conflict stops the cascade at that rung), plus per-cycle and rolling caps.
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Any, cast
+
+from roboco.config import settings
+from roboco.foundation import identity as _foundation
+from roboco.models.base import Complexity, TaskNature, TaskStatus, TaskType, Team
+from roboco.models.env_branches import ladder_pairs
+from roboco.services.base import BaseService
+from roboco.services.git import get_git_service
+from roboco.services.task import (
+ ENV_SYNC_SOURCE,
+ TaskCreateRequest,
+ TaskService,
+ get_task_service,
+)
+
+if TYPE_CHECKING:
+ from uuid import UUID
+
+ from sqlalchemy.ext.asyncio import AsyncSession
+
+ from roboco.db.tables import TaskTable
+
+
+class EnvSyncEngine(BaseService):
+ """Cascade the env ladder prod→…→head so dev never falls behind prod."""
+
+ service_name = "env_sync_engine"
+
+ def __init__(self, session: AsyncSession) -> None:
+ super().__init__(session)
+
+ async def run_cycle(self, projects: list[Any]) -> list[TaskTable]:
+ """Cascade each opted-in project's ladder; return conflict-PR tasks opened.
+
+ No-op unless ``env_sync_enabled``. Per-cycle + rolling caps; one open
+ env_sync task per repo (a conflict pauses the cascade at that rung).
+ Never pushes prod. Flushes; the caller (the orchestrator loop) owns the
+ commit.
+ """
+ if not settings.env_sync_enabled:
+ return []
+ task_svc = get_task_service(self.session)
+ open_count = len(await task_svc.list_open_env_sync_tasks())
+ created: list[TaskTable] = []
+ for project in projects:
+ if len(created) >= settings.env_sync_max_per_cycle:
+ break
+ if open_count >= settings.env_sync_max_open_tasks:
+ self.log.info(
+ "env-sync open-task cap reached; not cascading",
+ cap=settings.env_sync_max_open_tasks,
+ )
+ break
+ if not await self._should_sync(task_svc, project):
+ continue
+ task = await self._cascade_project(project)
+ if task is not None:
+ created.append(task)
+ open_count += 1
+ return created
+
+ async def _should_sync(self, task_svc: TaskService, project: Any) -> bool:
+ """True when ``project`` is opted in and not already being synced.
+
+ Opt-in = a declared env ladder (so there are pairs to cascade) + a git
+ token (the merges API needs a PAT). A degenerate ladder (head==prod)
+ has no pairs → skip. Dedup is per repo: a project with an open env_sync
+ task has its cascade paused at a conflicted rung, so skip it until the
+ sync PR resolves.
+ """
+ if project is None or getattr(project, "id", None) is None:
+ return False
+ # ProjectTable exposes ``git_token_encrypted`` (the column), not the
+ # ``has_git_token`` API-response boolean — the merges API needs a PAT.
+ if not getattr(project, "git_token_encrypted", None):
+ return False
+ if not ladder_pairs(project):
+ return False
+ existing = await task_svc.list_open_env_sync_tasks(
+ git_url=getattr(project, "git_url", None)
+ )
+ return not existing
+
+ async def _cascade_project(self, project: Any) -> TaskTable | None:
+ """Cascade one project top→down; on a conflict open one sync PR + task.
+
+ Clean / already-ancestor steps continue down the ladder. A missing ref
+ (misconfigured ladder or API error) skips the project without opening a
+ PR. A conflict opens a sync PR and a tracked task, then stops — never
+ cascade a dirty merge downward.
+ """
+ slug = str(getattr(project, "slug", "") or "")
+ git = get_git_service(self.session)
+ for upper, lower in ladder_pairs(project):
+ pair = f"{upper.branch}→{lower.branch}"
+ result = await git.sync_env_branch(slug, lower.branch, upper.branch)
+ status = result.get("status")
+ if status in ("merged", "already_ancestor"):
+ self.log.info("env-sync step", project=slug, pair=pair, status=status)
+ continue
+ if status == "missing_ref":
+ self.log.warning(
+ "env-sync step missing ref; skipping project",
+ project=slug,
+ pair=pair,
+ )
+ return None
+ # conflict: open a sync PR + tracked task, stop the cascade.
+ pr = await git.open_sync_pr(
+ slug,
+ upper.branch,
+ lower.branch,
+ body=(
+ f"The env-sync cascade could not merge `{upper.branch}` "
+ f"into `{lower.branch}` cleanly (non-fast-forward). Resolve "
+ f"the conflict and merge this PR so the cascade can resume.\n\n"
+ f"Opened automatically by the env-sync loop."
+ ),
+ )
+ if pr is None:
+ self.log.warning(
+ "env-sync conflict but sync PR open failed; skipping project",
+ project=slug,
+ pair=pair,
+ )
+ return None
+ task = await self._open_sync_task(project, upper.branch, lower.branch, pr)
+ self.log.info(
+ "env-sync conflict; sync PR opened",
+ project=slug,
+ pair=pair,
+ pr=pr["number"],
+ )
+ return task
+ return None
+
+ async def _open_sync_task(
+ self,
+ project: Any,
+ upper_branch: str,
+ lower_branch: str,
+ pr: dict[str, Any],
+ ) -> TaskTable:
+ """Open ONE PENDING, dispatchable coordination task tracking the sync PR."""
+ task_svc = get_task_service(self.session)
+ slug = str(getattr(project, "slug", "") or "")
+ return await task_svc.create(
+ TaskCreateRequest(
+ title=(
+ f"env-sync: resolve {upper_branch}→{lower_branch} conflict "
+ f"on {slug} (PR #{pr['number']})"
+ ),
+ description=(
+ f"The env-sync cascade could not merge `{upper_branch}` into "
+ f"`{lower_branch}` cleanly (conflict). A sync PR was opened:\n"
+ f"{pr['url']}\n\n"
+ "Resolve the conflict and merge the PR so the cascade can "
+ "resume. This is a Main-PM coordination root: decompose the "
+ "resolution and delegate the code work to a cell dev — the "
+ "Main PM does not resolve it directly. Opened automatically by "
+ "the env-sync loop; it still ships through the normal gates "
+ "(QA, PR review, and the CEO's merge)."
+ ),
+ acceptance_criteria=[
+ f"The sync PR {pr['url']} is merged (conflict resolved)",
+ f"`{lower_branch}` is not behind `{upper_branch}`",
+ ],
+ team=Team.MAIN_PM,
+ assigned_to=_foundation.AGENTS["main-pm"].uuid,
+ created_by=_foundation.AGENTS["system"].uuid,
+ task_type=TaskType.PLANNING,
+ nature=TaskNature.TECHNICAL,
+ estimated_complexity=Complexity.MEDIUM,
+ project_id=cast("UUID", project.id),
+ status=TaskStatus.PENDING,
+ source=ENV_SYNC_SOURCE,
+ confirmed_by_human=True,
+ )
+ )
+
+
+def get_env_sync_engine(session: AsyncSession) -> EnvSyncEngine:
+ """Construct an EnvSyncEngine bound to ``session``."""
+ return EnvSyncEngine(session)
diff --git a/roboco/services/git.py b/roboco/services/git.py
index ec4eaa3b..8c80155c 100644
--- a/roboco/services/git.py
+++ b/roboco/services/git.py
@@ -298,6 +298,10 @@ _REV_LIST_PARTS = 2
# GitHub REST API status codes
_GH_UNPROCESSABLE = 422
+# merges-API success codes: 201 = merge commit created + pushed; 204 = nothing
+# to merge (head already an ancestor of base).
+_HTTP_CREATED = 201
+_HTTP_NO_CONTENT = 204
# 404 means the PR (or repo) does not exist; surfaced as a typed GitError
# by `update_pr_for_task` so the gateway can convert it into a specific
# invalid_state envelope rather than the generic refusal message.
@@ -1122,9 +1126,9 @@ class GitService(BaseService):
parent = await task_service.get(UUID(str(task.parent_task_id)))
if parent and parent.branch_name:
return str(parent.branch_name)
- return await self._project_head_branch(project_slug)
+ return await self._project_default_branch(project_slug)
- async def _project_head_branch(self, project_slug: str) -> str:
+ async def _project_default_branch(self, project_slug: str) -> str:
"""Return the project's head environment branch (ladder index 0).
This is where dev/cell/leaf PRs target — the dev trunk. Falls back to
@@ -1196,7 +1200,7 @@ class GitService(BaseService):
base_branch = await self._resolve_base_branch(
task_id, request.parent_branch, request.project_slug, task_service
)
- default_branch = await self._project_head_branch(request.project_slug)
+ default_branch = await self._project_default_branch(request.project_slug)
# Token for any remote-touching git command below (fetch, ls-remote,
# pull, push). Injected into a single `http.extraheader` config for
@@ -2422,7 +2426,7 @@ class GitService(BaseService):
Returns: (pr_number, pr_url, title, source_branch, target_branch)
"""
source_branch = await self._pr_head_branch(workspace, request)
- default_branch = await self._project_head_branch(request.project_slug)
+ default_branch = await self._project_default_branch(request.project_slug)
git_token = await self._get_project_token_or_raise(request.project_slug)
target_branch, pr_title, pr_body = await self._resolve_new_pr_context(
workspace, request, source_branch, default_branch, git_token
@@ -2463,6 +2467,154 @@ class GitService(BaseService):
target_branch,
)
+ async def sync_env_branch(
+ self, project_slug: str, target_branch: str, source_branch: str
+ ) -> dict[str, Any]:
+ """Merge ``source_branch`` (an upper env rung) into ``target_branch`` (the
+ lower rung) server-side via GitHub's merges API — one step of the
+ prod→head cascade. The merge commit lands on ``target_branch`` (the
+ clean-cascade auto-push). The cascade's target is never prod by
+ construction (``ladder_pairs``), so prod is never pushed here.
+
+ Returns ``{"status": ...}``:
+
+ * ``already_ancestor`` — target already contains source (HTTP 204).
+ * ``merged`` — merge commit created + pushed to target (HTTP 201; ``sha``).
+ * ``conflict`` — non-fast-forward / merge conflict (HTTP 409); no commit.
+ * ``missing_ref`` — no token / unparseable remote / a branch absent (422).
+
+ Never raises into the engine loop. Does NOT open a PR on conflict —
+ the caller decides that.
+ """
+ project = await get_project_service(self.session).get_by_slug(project_slug)
+ if project is None or not project.git_url:
+ return {"status": "missing_ref"}
+ git_token = await self._token_for_project(project_slug)
+ if not git_token:
+ return {"status": "missing_ref"}
+ try:
+ owner, repo = self._parse_git_url(project.git_url)
+ except GitError:
+ return {"status": "missing_ref"}
+ try:
+ async with httpx.AsyncClient(timeout=_default_git_timeout()) as client:
+ resp = await client.post(
+ f"{_api_base()}/repos/{owner}/{repo}/merges",
+ headers={
+ "Authorization": f"Bearer {git_token}",
+ "Accept": "application/vnd.github+json",
+ "X-GitHub-Api-Version": "2022-11-28",
+ },
+ json={
+ "base": target_branch,
+ "head": source_branch,
+ "commit_message": f"sync: {source_branch} → {target_branch}",
+ },
+ )
+ except httpx.HTTPError as exc:
+ self.log.warning(
+ "env-sync merges API error", project=project_slug, error=str(exc)
+ )
+ return {"status": "missing_ref"}
+ return self._env_merge_status(resp, project_slug)
+
+ def _env_merge_status(
+ self, resp: httpx.Response, project_slug: str
+ ) -> dict[str, Any]:
+ """Map a GitHub merges-API response to an env-sync status dict.
+
+ ``merged`` carries the new merge ``sha``; ``conflict`` (409) leaves the
+ target untouched; any other code (incl. 422 missing-ref / no-merge) is
+ ``missing_ref`` so the engine skips without opening a PR.
+ """
+ if resp.status_code == _HTTP_CREATED:
+ return {"status": "merged", "sha": resp.json().get("sha")}
+ if resp.status_code == _HTTP_NO_CONTENT:
+ return {"status": "already_ancestor"}
+ if resp.status_code == _HTTP_CONFLICT:
+ return {"status": "conflict"}
+ self.log.warning(
+ "env-sync merges API unexpected status",
+ project=project_slug,
+ status=resp.status_code,
+ body=resp.text[:200],
+ )
+ return {"status": "missing_ref"}
+
+ async def open_sync_pr(
+ self, project_slug: str, source_branch: str, target_branch: str, body: str
+ ) -> dict[str, Any] | None:
+ """Open (or reuse) a sync PR ``source_branch → target_branch``.
+
+ Idempotent: reuses an already-open PR for the same head→base. Returns
+ ``{"number", "url"}`` or None on a missing token / unparseable remote /
+ GitHub error — never raises into the engine loop.
+ """
+ project = await get_project_service(self.session).get_by_slug(project_slug)
+ if project is None or not project.git_url:
+ return None
+ git_token = await self._token_for_project(project_slug)
+ if not git_token:
+ return None
+ try:
+ owner_repo = self._parse_git_url(project.git_url)
+ except GitError:
+ return None
+ return await self._post_sync_pr(
+ owner_repo, git_token, (source_branch, target_branch), body, project_slug
+ )
+
+ async def _post_sync_pr(
+ self,
+ owner_repo: tuple[str, str],
+ git_token: str,
+ branches: tuple[str, str],
+ body: str,
+ project_slug: str,
+ ) -> dict[str, Any] | None:
+ """Reuse an open sync PR or create a new one for ``source→target``.
+
+ Never raises into the engine loop: a missing existing PR, a rejected
+ create, or a transport error all return None.
+ """
+ owner, repo = owner_repo
+ source_branch, target_branch = branches
+ existing = await self._find_existing_pr(
+ owner, repo, source_branch, target_branch, git_token
+ )
+ if existing is not None:
+ return {
+ "number": int(existing["number"]),
+ "url": str(existing.get("html_url", "")),
+ }
+ try:
+ resp = await self._post_pr(
+ owner,
+ repo,
+ git_token,
+ {
+ "title": f"sync: {source_branch} → {target_branch}",
+ "body": body,
+ "head": source_branch,
+ "base": target_branch,
+ },
+ )
+ except GitError as exc:
+ self.log.warning(
+ "env-sync PR create failed", project=project_slug, error=str(exc)
+ )
+ return None
+ if not resp.is_success:
+ self.log.warning(
+ "env-sync PR create rejected",
+ project=project_slug,
+ status=resp.status_code,
+ body=resp.text[:200],
+ )
+ return None
+ data = resp.json()
+ return {"number": int(data["number"]), "url": str(data.get("html_url", ""))}
+
async def _resolve_new_pr_context(
self,
workspace: Path,
@@ -3514,7 +3666,7 @@ class GitService(BaseService):
await self._delete_pr_branch_best_effort(owner, repo, pr_number, git_token)
- target_branch = await self._project_head_branch(project_slug)
+ target_branch = await self._project_default_branch(project_slug)
# Default branch always exists on origin, so the plain sync is correct
# here. The best-effort variant guards the agent-facing pr_merge path,
# whose target can be an integration branch deleted from origin.
@@ -3930,7 +4082,7 @@ class GitService(BaseService):
layering is preserved. Fall back to the default branch only if the
create push itself fails.
"""
- default_branch = await self._project_head_branch(project_slug)
+ default_branch = await self._project_default_branch(project_slug)
if base_branch == default_branch:
return base_branch
ls = await self._run_git(
@@ -4296,7 +4448,7 @@ class GitService(BaseService):
# repo's default branch — a root→master PR is merged solely by the CEO
# via approve-&-merge (merge_pr_for_task, CEO-gated from
# awaiting_ceo_approval). Agents open the master PR and escalate.
- default_branch = await self._project_head_branch(project.slug)
+ default_branch = await self._project_default_branch(project.slug)
if target == default_branch:
raise UnauthorizedError(
action="pr_merge",
diff --git a/roboco/services/release_executor.py b/roboco/services/release_executor.py
index d361b45b..77dc374f 100644
--- a/roboco/services/release_executor.py
+++ b/roboco/services/release_executor.py
@@ -527,7 +527,7 @@ def _resolve_release_ci_workflow() -> str:
async def get_release_executor(session: AsyncSession) -> ReleaseExecutor:
"""Build a ReleaseExecutor with a production ops over a fresh writable clone."""
from roboco.config import settings
- from roboco.models.env_branches import effective_environments, prod_branch
+ from roboco.models.env_branches import prod_branch, promotion_chain
from roboco.services.project import get_project_service
slug = (settings.self_heal_project_slug or "roboco-api").strip()
@@ -541,10 +541,8 @@ async def get_release_executor(session: AsyncSession) -> ReleaseExecutor:
# of the dev retarget. default_branch stays as the legacy/shim source.
default_branch = prod_branch(project)
# Full-chain promotion: rung branches head→…→just-below-prod to merge into
- # the prod checkout before bumping. Skips the prod rung itself and any rung
- # sharing the prod branch (degenerate head==prod → empty → no-op).
- rungs = effective_environments(project)
- env_chain = [r.branch for r in rungs[:-1] if r.branch != default_branch]
+ # the prod checkout before bumping. Degenerate (head==prod) → [] → no-op.
+ env_chain = promotion_chain(project)
# PAT rides a per-call ``-c http.extraheader=Authorization: Basic …`` config
# (mirrors workspace.py :642 / :1285) so it never lands in the clone/push
# argv — ``/proc//cmdline`` would otherwise expose a URL-embedded token.
diff --git a/roboco/services/release_manager_engine.py b/roboco/services/release_manager_engine.py
index eeb10e48..4452cebd 100644
--- a/roboco/services/release_manager_engine.py
+++ b/roboco/services/release_manager_engine.py
@@ -37,6 +37,7 @@ from roboco.services.notification import NotificationService
from roboco.services.project import get_project_service
from roboco.services.release_readiness import (
ReleaseReadinessReport,
+ _run_git,
assess,
gather_snapshot,
report_to_dict,
@@ -213,7 +214,6 @@ class ReleaseManagerEngine(BaseService):
"""
from roboco.models.env_branches import head_branch, prod_branch
from roboco.services.git import get_git_service
- from roboco.services.release_readiness import _run_git
from roboco.services.workspace import get_workspace_service
slug = _roboco_slug()
@@ -250,15 +250,9 @@ class ReleaseManagerEngine(BaseService):
# already present. Best-effort: a fetch failure degrades to last_tag..HEAD
# (gather_snapshot falls back when prod_tip is None) — never aborts.
prod_name = prod_branch(project)
- have_prod = prod_name != head_branch(project)
- if have_prod:
- try:
- await asyncio.to_thread(
- _run_git, Path(root), ["fetch", "origin", prod_name]
- )
- except Exception as exc:
- self.log.warning("release-manager: prod fetch failed", error=str(exc))
- have_prod = False
+ prod_for_snapshot = await self._ensure_prod_fetched(
+ root, prod_name, prod_name != head_branch(project)
+ )
today = datetime.now(UTC).strftime("%Y-%m-%d")
# gather_snapshot runs multiple sync `subprocess.run` git calls + a
# filesystem walk; offload so the shared API event loop isn't blocked
@@ -267,10 +261,30 @@ class ReleaseManagerEngine(BaseService):
gather_snapshot,
Path(root),
master_ci_conclusion=conclusion,
- prod_branch=prod_name if have_prod else None,
+ prod_branch=prod_for_snapshot,
)
return assess(snapshot, today=today)
+ async def _ensure_prod_fetched(
+ self, root: Path, prod_name: str, needs_fetch: bool
+ ) -> str | None:
+ """Ensure ``origin/`` is fetched for the prod..head readiness diff.
+
+ Degenerate (prod==head) needs no fetch — the read clone is pinned to
+ head==prod. Best-effort: a fetch failure returns None so gather_snapshot
+ falls back to last_tag..HEAD rather than aborting the assessment.
+ """
+ if not needs_fetch:
+ return prod_name
+ try:
+ await asyncio.to_thread(
+ _run_git, Path(root), ["fetch", "origin", prod_name]
+ )
+ except Exception as exc:
+ self.log.warning("release-manager: prod fetch failed", error=str(exc))
+ return None
+ return prod_name
+
def get_release_manager_engine(
session: AsyncSession, assessor: ReleaseAssessor | None = None
diff --git a/roboco/services/settings.py b/roboco/services/settings.py
index 418abfaf..b0f55cf0 100644
--- a/roboco/services/settings.py
+++ b/roboco/services/settings.py
@@ -62,6 +62,7 @@ FEATURE_FLAGS: tuple[tuple[str, str], ...] = (
("gateway_health_enabled", "Gateway-health recovery"),
("ci_watch_enabled", "Multi-repo CI-watch"),
("dep_update_enabled", "Dependency-update bot"),
+ ("env_sync_enabled", "Environment-branch sync (cascade prod→dev)"),
("docs_sync_enabled", "Docs-divergence sync (release -> docs-update task)"),
("release_manager_enabled", "Gated release manager"),
("org_memory_enabled", "Organizational memory loop"),
diff --git a/roboco/services/task.py b/roboco/services/task.py
index af69fdb1..2c00bf9c 100644
--- a/roboco/services/task.py
+++ b/roboco/services/task.py
@@ -603,6 +603,12 @@ DEP_UPDATE_SOURCE = "dep_update"
# auto-merges; requires the roboco-website project to be registered.
DOCS_SYNC_SOURCE = "docs_sync"
+# Source tag for an env-sync conflict task: opened by the EnvSyncEngine when the
+# prod→…→head cascade hits a non-fast-forward on a rung and opens a sync PR. Rides
+# the normal delivery lifecycle (+ PR-review gate) and is never auto-merged; the
+# task tracks the PR so the dedup cap + panel visibility work.
+ENV_SYNC_SOURCE = "env_sync"
+
# Source tag for a gated release proposal: opened by the release-manager engine
# when accumulated unreleased changes pass the threshold + the gate is green.
# Unlike the sources above it is NEVER dispatched — it is HELD for the CEO
@@ -1575,6 +1581,26 @@ class TaskService(BaseService):
result = await self.session.execute(stmt)
return list(result.scalars().all())
+ async def list_open_env_sync_tasks(
+ self, git_url: str | None = None
+ ) -> list[TaskTable]:
+ """Non-terminal env_sync conflict tasks — the dedupe + open-cap basis.
+
+ Optionally scoped to one repo by ``git_url`` (matched on the normalized
+ repo key). A conflict stops the cascade at that rung, so at most one
+ open env_sync task exists per repo: while it is open the repo's cascade
+ is paused and the engine skips it until the PR resolves.
+ """
+ stmt = select(TaskTable).where(
+ TaskTable.source == ENV_SYNC_SOURCE,
+ TaskTable.status.notin_([TaskStatus.COMPLETED, TaskStatus.CANCELLED]),
+ )
+ if git_url is not None:
+ stmt = stmt.join(ProjectTable, TaskTable.project_id == ProjectTable.id)
+ stmt = stmt.where(_repo_key_expr(ProjectTable.git_url) == repo_key(git_url))
+ result = await self.session.execute(stmt)
+ return list(result.scalars().all())
+
async def list_open_docs_sync_tasks(
self, version: str | None = None
) -> list[TaskTable]:
diff --git a/tests/e2e_smoke/test_background_engines.py b/tests/e2e_smoke/test_background_engines.py
index beb1787d..f172b9f0 100644
--- a/tests/e2e_smoke/test_background_engines.py
+++ b/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"
diff --git a/tests/integration/services/test_env_sync_engine.py b/tests/integration/services/test_env_sync_engine.py
new file mode 100644
index 00000000..1c0d8bb2
--- /dev/null
+++ b/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 == []
diff --git a/tests/integration/test_migration_env_branches.py b/tests/integration/test_migration_env_branches.py
new file mode 100644
index 00000000..f4a0fa77
--- /dev/null
+++ b/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"},
+ ]
diff --git a/tests/unit/models/test_env_branches.py b/tests/unit/models/test_env_branches.py
new file mode 100644
index 00000000..347867dc
--- /dev/null
+++ b/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"}]
diff --git a/tests/unit/runtime/test_orchestrator_shutdown_drain.py b/tests/unit/runtime/test_orchestrator_shutdown_drain.py
index b1a7dfb7..d327c859 100644
--- a/tests/unit/runtime/test_orchestrator_shutdown_drain.py
+++ b/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",
diff --git a/tests/unit/services/test_release_executor.py b/tests/unit/services/test_release_executor.py
index 161e9483..04b2f98d 100644
--- a/tests/unit/services/test_release_executor.py
+++ b/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")
diff --git a/tests/unit/services/test_release_executor_commit_fail_closed.py b/tests/unit/services/test_release_executor_commit_fail_closed.py
index f3d8a34f..e24fc29d 100644
--- a/tests/unit/services/test_release_executor_commit_fail_closed.py
+++ b/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=[],
)