feat(merge-gate): auto-rebase onto current main + re-test + serialise merges

Deterministic (no-LLM) sub-gate on the deploy-staging -> deploy edge that
catches a feature branch up to the CURRENT origin/main, re-tests the combined
tree, and serialises merges with a per-repo file lease — so two green parallel
branches can no longer break main (self-hosting safety for the orchestrator repo).

- src/merge_gate.py: branch_is_behind_main, auto_rebase_onto_main (push
  --force-with-lease ONLY the task branch, NEVER main), retest_branch, and a
  file merge-lease (atomic O_CREAT|O_EXCL, holder-aware release, stale reclaim).
  Strict never-raise contract; all git ops in the per-branch worktree.
- src/qg/checks.py: check_branch_mergeable composes the primitives under the
  lease; registered in QG_CHECKS. Conditional rollout (merge_gate_enabled /
  merge_gate_repos, default self-hosting only).
- src/stage_engine.py: sub-gate hook on deploy-staging (not a new stage). PASS ->
  advance; "merge-lock busy" -> DEFER (re-queue with available_at, anti-deadlock
  at max_concurrency=1, capped); conflict/red re-test -> rollback to development
  + developer retry (capped by MAX_DEVELOPER_RETRIES). Lease released on
  deploy->done / rollback / PR-merged webhook.
- src/db.py: enqueue_job(available_at_delay_s=...) for the defer (no schema change).
- src/webhooks/gitea.py: holder-aware lease release on PR-merged.
- src/config.py + .env.example: ORCH_MERGE_* settings.

Docs: README + adr-0006 (architect) already cover the design; CHANGELOG updated.
Tests: test_merge_gate.py, test_qg_merge_gate.py, test_merge_gate_race.py,
test_stage_engine.py::TestMergeGate, test_config.py, QG-registry snapshot.
Full suite: 535 passed.

Refs: ORCH-043

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

.env.example

@@ -17,3 +17,22 @@ ORCH_DB_PATH=/app/data/orchestrator.db
 # one is sent silently to the BOTTOM of the chat (deleteMessage + sendMessage +
 # repoint). One card per task in both modes. Any value other than "bump" -> edit.
 ORCH_TRACKER_MODE=edit
+# ORCH-043: merge-gate (auto-rebase onto current origin/main + re-test + merge-lock)
+# on the deploy-staging -> deploy edge. Deterministic sub-gate (no LLM) that catches
+# the branch up to the CURRENT origin/main, re-tests it, and serialises merges so two
+# green parallel branches can't break main.
+#   ENABLED   -> global kill-switch (false -> whole gate is a no-op pass).
+#   REPOS     -> CSV of repos where the gate is REAL; empty -> only the self-hosting
+#                repo (orchestrator); other repos -> conditional no-op (mirrors ORCH-35).
+#   RETEST_TIMEOUT_S -> wall-clock budget for the post-rebase re-test.
+#   RETEST_TARGET    -> pytest target for the re-test.
+#   LOCK_TIMEOUT_S   -> max merge-lease age before a stale lease is reclaimed.
+#   DEFER_DELAY_S    -> delay before re-running the gate when the lock is busy.
+#   DEFER_MAX_ATTEMPTS -> defer retries before escalation (avoids livelock).
+ORCH_MERGE_GATE_ENABLED=true
+ORCH_MERGE_GATE_REPOS=
+ORCH_MERGE_RETEST_TIMEOUT_S=600
+ORCH_MERGE_RETEST_TARGET=tests/
+ORCH_MERGE_LOCK_TIMEOUT_S=300
+ORCH_MERGE_DEFER_DELAY_S=60
+ORCH_MERGE_DEFER_MAX_ATTEMPTS=5

CHANGELOG.md

@@ -5,6 +5,7 @@
 ## [Unreleased]
 
 ### Added
+- **Merge-gate: авто-rebase на текущий `origin/main` + повторный прогон тестов + сериализация мержей** (ORCH-043): детерминированный (без LLM) суб-гейт на ребре `deploy-staging → deploy`, выполняемый ПЕРЕД мержем PR деплоером. Закрывает класс гонок «две зелёные ветки в одном репо ломают `main`»: пайплайн валидирует ветку против того `main`, от которого она ответвилась, а не против `main` в момент мержа — между «ветка зелёная» и «ветка смержена» параллельная задача может сдвинуть `main` (семантический конфликт: git мержит без текстового конфликта, но совмещённый `main` красный). Для self-hosting репозитория `orchestrator` это означало бы красный `main` инструмента, обслуживающего ВСЕ проекты. Новый модуль `src/merge_gate.py` (контракт «never raise», все git-операции — в per-branch worktree, ORCH-2/S-4): `branch_is_behind_main` (`git merge-base --is-ancestor origin/main HEAD`), `auto_rebase_onto_main` (rebase + `git push --force-with-lease` ТОЛЬКО ветки задачи — `main` НИКОГДА не пушится; текстовый конфликт → `rebase --abort` + чистый worktree), `retest_branch` (`python -m pytest <target>` в догнанном worktree, бюджет `merge_retest_timeout_s`), файловый merge-lease (`acquire_merge_lease`/`release_merge_lease`, атомарный `O_CREAT|O_EXCL`, holder-aware release, реклейм протухшего/битого лиза — без изменения схемы БД). Новый quality-gate `check_branch_mergeable` (`src/qg/checks.py`, зарегистрирован в `QG_CHECKS`) композирует примитивы под лизом: kill-switch/вне-области → no-op pass; lock занят → `(False, "merge-lock busy")` (сигнал DEFER, не код-фолт); ветка свежая → pass (лиз ДЕРЖИТСЯ до мержа); отстала → rebase → конфликт = fail+release, чисто → retest → зелёный = pass (лиз держится) / красный|timeout = fail+release. Интеграция в `src/stage_engine.py` (суб-гейт на `deploy-staging`, БЕЗ новой стадии в `STAGE_TRANSITIONS`): pass → advance на `deploy`; «merge-lock busy» → DEFER (повторная постановка деплоера на `deploy-staging` с задержкой `available_at`, анти-дедлок при `max_concurrency=1`, restart-safe счётчик по `task_content`, лимит `merge_defer_max_attempts` → block+Telegram); конфликт/красный retest → ROLLBACK на `development` + ретрай developer-а (кап `MAX_DEVELOPER_RETRIES`, без бесконечного баунса). Лиз освобождается на `deploy→done`, на rollback и по webhook смерженного PR (`src/webhooks/gitea.py`). Новый параметр `enqueue_job(..., available_at_delay_s=...)` (`src/db.py`) — отложенная постановка без изменения схемы. Условность раскатки (зеркало ORCH-35): `merge_gate_repos` (CSV) или по умолчанию только self-hosting `orchestrator`; глобальный kill-switch `merge_gate_enabled`. Новые настройки `ORCH_MERGE_GATE_ENABLED` (true), `ORCH_MERGE_GATE_REPOS` (""), `ORCH_MERGE_RETEST_TIMEOUT_S` (600), `ORCH_MERGE_RETEST_TARGET` (tests/), `ORCH_MERGE_LOCK_TIMEOUT_S` (300), `ORCH_MERGE_DEFER_DELAY_S` (60), `ORCH_MERGE_DEFER_MAX_ATTEMPTS` (5). ADR `docs/work-items/ORCH-043/06-adr/ADR-001-merge-gate.md`, глобальный `docs/architecture/adr/adr-0006-merge-gate.md`. Тесты: `tests/test_merge_gate.py`, `tests/test_qg_merge_gate.py`, `tests/test_merge_gate_race.py`, `tests/test_stage_engine.py::TestMergeGate`, `tests/test_config.py`.
 - **Режим `bump` live-трекера Telegram** (ORCH-042): новый `ORCH_TRACKER_MODE` (`Settings.tracker_mode`, дефолт `edit`) выбирает поведение карточки задачи. `edit` (как было) — карточка редактируется на месте (`editMessageText`). `bump` — на каждом обновлении старое сообщение удаляется и карточка отправляется заново вниз чата (best-effort `delete_telegram(старый_id)` → `send_telegram(text, disable_notification=True)` → `set_tracker_message_id(new_id)`), чтобы актуальный статус всегда был последним в чате при активной переписке. Инвариант «одна карточка на задачу» сохранён в обоих режимах: за один вызов `update_task_tracker` шлётся ≤1 нового сообщения; `set_tracker_message_id` вызывается ТОЛЬКО при успешном send (транзиентный `None` не затирает указатель); результат delete НЕ блокирует отправку новой карточки (delete-fail у сообщения >48ч → всё равно шлём новое). Резолюция режима в `notifications` (case-insensitive, trim): всё, что ≠ `"bump"` (включая пустое/мусор) → `edit` → нулевая регрессия и оркестратор не падает на любом значении флага. Новый low-level helper `delete_telegram(message_id) -> bool` (контракт «never raises», маркеры `_DELETE_GONE_MARKERS`): `ok:true` или «уже нет / нельзя удалить» → `True`; неизвестный `ok:false`/5xx/исключение → `False`; нет кредов → `False` без HTTP. Сигнатуры `send_telegram`/`edit_telegram`/`update_task_tracker` и схема БД (`tasks.tracker_message_id`) не менялись. ADR `docs/work-items/ORCH-042/06-adr/ADR-001-tracker-bump-mode.md`. Тесты: `tests/test_tracker_bump.py`, `tests/test_config.py`.
 - **Дословный текст findings reviewer/tester встраивается в `task_desc` заворота** (ORCH-046): при откате на `development` строка `task_desc` (попадает в `.task-dev.md` developer-агента) теперь несёт суть претензий, а не только ссылку на файл — устраняет «испорченный телефон», из-за которого агент шёл «читать файл», терял ключевые P0/P1 / причину FAIL и заворачивался снова, выжигая `MAX_DEVELOPER_RETRIES` и токены. Новый defensive-модуль `src/review_parse.py` (контракт «never raise», как `src/frontmatter.py`): `extract_review_findings(path)` — дословные пункты P0/P1 из секции `## Findings` файла `12-review.md`; `extract_test_failures(path)` — релевантный фрагмент тела `13-test-report.md` (приоритет `## Вывод pytest` → FAIL-строки `## Результаты` → `## Итог`). Обе функции усекают результат до `MAX_FINDINGS_CHARS`/`MAX_FAILURES_CHARS` (≈2000) с маркером `…(truncated)`. Две rollback-ветки `src/stage_engine.py` (reviewer REQUEST_CHANGES, tester `check_tests_passed` FAIL) встраивают извлечённый текст и **сохраняют ссылку** на полный файл («Полный контекст»); при пустом/битом артефакте — graceful-фоллбэк на прежнюю ссылку-строку (никаких исключений в `advance_stage`). Tester-ветка дополнительно всегда включает `reason` гейта. Последовательность отката, `_developer_retry_count`, поля `AdvanceResult` и реестр `QG_CHECKS` не менялись. ADR `docs/work-items/ORCH-046/06-adr/ADR-001-embed-findings-in-task-desc.md`. Тесты: `tests/test_review_parse.py`, `tests/test_stage_engine.py::TestRollbackTaskDescEmbedding`.
 - **Поллинг с ретраем в quality-gate `check_ci_green`** (ORCH-045): гейт CI превращён из single-shot в polling, чтобы устранить race condition — раньше один опрос combined commit-status сразу после пуша developer-а ловил транзиентный `pending` (типично 1-3с, реальный кейс ORCH-017: опрос 17:58:54 → pending, CI дозеленел 17:58:55) и задача застревала насмерть без повторного опроса. Теперь: `success` → пропуск сразу; `failure`/`error` → провал сразу (терминально, ретрай бессмыслен); `pending`/unknown → `time.sleep` и повторный опрос до `ci_poll_max_attempts` раз; истечение попыток → явный `(False, "CI still pending after <T>s")` (тупик больше не молчаливый); 404 → как раньше; транзиентная `httpx.HTTPError` на попытке логируется и ретраится в рамках бюджета. Параметры — новые настройки `ORCH_CI_POLL_MAX_ATTEMPTS` (12) и `ORCH_CI_POLL_INTERVAL_S` (10) в `src/config.py` (~2 мин ожидания pending). Сигнатура `check_ci_green(repo, branch)` и реестр `QG_CHECKS` не менялись; `check_tests_passed` не затронут. ADR `docs/architecture/adr/adr-0004-ci-poll-retry.md`. Тесты: `tests/test_qg.py::TestCheckCIGreen`.

src/config.py

@@ -130,6 +130,28 @@ class Settings(BaseSettings):
     ci_poll_max_attempts: int = 12
     ci_poll_interval_s: int = 10
 
+    # ORCH-043: merge-gate (auto-rebase + re-test + merge-lock) on the
+    # deploy-staging -> deploy edge. A deterministic sub-gate (no LLM) that
+    # catches the up-to-date branch up to the CURRENT origin/main, re-tests it,
+    # and serialises merges so two green branches can't break main.
+    #   merge_gate_enabled     -> global kill-switch; False -> no-op pass for the
+    #                             whole gate (staged rollout, env ORCH_MERGE_GATE_ENABLED).
+    #   merge_gate_repos       -> CSV of repos where the gate is REAL; empty means
+    #                             only the self-hosting repo (orchestrator). Other
+    #                             repos -> conditional no-op (mirrors ORCH-35 staging).
+    #   merge_retest_timeout_s -> wall-clock budget for the post-rebase re-test.
+    #   merge_retest_target    -> pytest target for the re-test (portability across repos).
+    #   merge_lock_timeout_s   -> max lease age; an older lease is reclaimed (crash backstop).
+    #   merge_defer_delay_s    -> delay before re-running the gate when the lock is busy.
+    #   merge_defer_max_attempts -> defer retries before escalation (avoids livelock).
+    merge_gate_enabled: bool = True
+    merge_gate_repos: str = ""
+    merge_retest_timeout_s: int = 600
+    merge_retest_target: str = "tests/"
+    merge_lock_timeout_s: int = 300
+    merge_defer_delay_s: int = 60
+    merge_defer_max_attempts: int = 5
+
     # Telegram notifications
     telegram_bot_token: str = ""
     telegram_chat_id: str = ""

src/db.py

@@ -324,19 +324,34 @@ def enqueue_job(
     task_content: str | None = None,
     task_id: int | None = None,
     max_attempts: int = 2,
+    available_at_delay_s: int | None = None,
 ) -> int:
     """Enqueue a new job (status='queued'). Returns the new job id.
 
     This is what webhook handlers call instead of launching an agent in-process:
     it is a fast DB INSERT that returns immediately. The background worker
     (queue_worker) picks the job up later.
+
+    ORCH-043 (merge-gate defer): when ``available_at_delay_s`` is given the job's
+    ``available_at`` is set to ``now + delay`` so claim_next_job won't pick it up
+    until the delay elapses (re-uses the existing ORCH-1 backoff gate). Used to
+    re-queue the staging-deployer after a "merge-lock busy" defer without burning a
+    worker slot in a blocking wait.
     """
     conn = get_db()
-    cursor = conn.execute(
-        "INSERT INTO jobs (agent, repo, task_id, task_content, max_attempts) "
-        "VALUES (?, ?, ?, ?, ?)",
-        (agent, repo, task_id, task_content, max_attempts),
-    )
+    if available_at_delay_s is not None:
+        cursor = conn.execute(
+            "INSERT INTO jobs (agent, repo, task_id, task_content, max_attempts, available_at) "
+            "VALUES (?, ?, ?, ?, ?, datetime('now', ?))",
+            (agent, repo, task_id, task_content, max_attempts,
+             f"+{int(available_at_delay_s)} seconds"),
+        )
+    else:
+        cursor = conn.execute(
+            "INSERT INTO jobs (agent, repo, task_id, task_content, max_attempts) "
+            "VALUES (?, ?, ?, ?, ?)",
+            (agent, repo, task_id, task_content, max_attempts),
+        )
     job_id = cursor.lastrowid
     conn.commit()
     conn.close()

src/merge_gate.py

@@ -0,0 +1,340 @@
+"""Merge-gate core (ORCH-043): catch a branch up to the CURRENT origin/main,
+re-test it, and serialise merges with a file lease.
+
+Background
+----------
+The pipeline validates a branch against the ``main`` it was BRANCHED from, not the
+``main`` at the moment of merge. Between "branch validated" and "branch merged" a
+parallel task may have advanced ``main`` -> a *semantic* merge conflict: git merges
+with no textual conflict, yet the combined ``main`` is broken. For the self-hosting
+``orchestrator`` repo that means a red ``main`` of the tool serving every project.
+
+This module provides the deterministic (no-LLM) primitives the quality-gate
+``check_branch_mergeable`` (src/qg/checks.py) composes on the
+``deploy-staging -> deploy`` edge, BEFORE the deployer merges the PR:
+
+  * ``branch_is_behind_main``   -> is the branch missing the latest origin/main?
+  * ``auto_rebase_onto_main``   -> rebase onto origin/main + push --force-with-lease
+                                   (ONLY the task branch; NEVER main).
+  * ``retest_branch``           -> run the project test-suite in the caught-up worktree.
+  * file lease (``acquire_merge_lease`` / ``release_merge_lease``) -> serialise the
+    "catch-up + re-test + merge" of ONE repo, held from the gate to the actual merge.
+
+Invariants (self-hosting safety, ТЗ §10):
+  * NEVER push or force-push ``main`` — the only force op is ``--force-with-lease``
+    on the task branch.
+  * All git ops run in the per-branch worktree (ensure_worktree), never the shared clone.
+  * Every public function honours a strict **never-raise** contract: any git/OS error
+    -> ``(False, "<reason>")`` (or a safe bool), never a propagated exception.
+"""
+
+import json
+import logging
+import os
+import subprocess
+import time
+
+from .config import settings
+from .git_worktree import ensure_worktree, get_worktree_path
+
+logger = logging.getLogger("orchestrator.merge_gate")
+
+# git sub-command timeouts (seconds). Generous but bounded so a hung git never
+# wedges the monitor-thread that runs the gate.
+_FETCH_TIMEOUT = 60
+_REBASE_TIMEOUT = 120
+_PUSH_TIMEOUT = 60
+_SHORT_TIMEOUT = 30
+
+
+# ---------------------------------------------------------------------------
+# behind / ancestor detection
+# ---------------------------------------------------------------------------
+def branch_is_behind_main(repo: str, branch: str) -> bool:
+    """Return True iff ``branch`` does NOT already contain the latest origin/main.
+
+    A branch is "behind" when ``origin/main`` is **not** an ancestor of the branch
+    HEAD (``git merge-base --is-ancestor origin/main HEAD`` returns non-zero). All
+    work happens in the per-branch worktree (ORCH-2 / S-4 isolation).
+
+    Never-raise (AC-9 / TC-03): any git/OS failure or an ambiguous result is treated
+    as "cannot prove the branch is up-to-date" -> return True (force a rebase attempt
+    rather than merge blindly). It returns a bool, never raises.
+    """
+    try:
+        wt = ensure_worktree(repo, branch)
+    except Exception as e:  # noqa: BLE001 - never-raise contract
+        logger.warning("branch_is_behind_main: worktree error for %s/%s: %s", repo, branch, e)
+        return True
+
+    try:
+        subprocess.run(
+            ["git", "-C", wt, "fetch", "origin", "main"],
+            capture_output=True, timeout=_FETCH_TIMEOUT,
+        )
+        r = subprocess.run(
+            ["git", "-C", wt, "merge-base", "--is-ancestor", "origin/main", "HEAD"],
+            capture_output=True, timeout=_SHORT_TIMEOUT,
+        )
+    except (subprocess.SubprocessError, OSError) as e:
+        logger.warning("branch_is_behind_main: git error for %s/%s: %s", repo, branch, e)
+        return True
+
+    if r.returncode == 0:
+        # origin/main IS an ancestor of HEAD -> branch already up-to-date.
+        return False
+    if r.returncode == 1:
+        # origin/main is NOT an ancestor -> branch is behind.
+        return True
+    # Any other code (e.g. bad ref) -> ambiguous; do not merge blindly.
+    logger.warning(
+        "branch_is_behind_main: ambiguous merge-base rc=%s for %s/%s (treating as behind)",
+        r.returncode, repo, branch,
+    )
+    return True
+
+
+def _conflicted_files(wt: str) -> str:
+    """Best-effort list of unmerged (conflicting) files in the worktree."""
+    try:
+        r = subprocess.run(
+            ["git", "-C", wt, "diff", "--name-only", "--diff-filter=U"],
+            capture_output=True, text=True, timeout=_SHORT_TIMEOUT,
+        )
+        files = r.stdout.strip().replace("\n", ", ")
+        return files or "unknown"
+    except (subprocess.SubprocessError, OSError):
+        return "unknown"
+
+
+# ---------------------------------------------------------------------------
+# auto-rebase onto origin/main
+# ---------------------------------------------------------------------------
+def auto_rebase_onto_main(repo: str, branch: str) -> tuple[bool, str]:
+    """Catch ``branch`` up to ``origin/main`` via rebase, then push it.
+
+    Steps (all in the per-branch worktree):
+      1. ``git fetch origin main``.
+      2. ``git rebase origin/main``:
+         - textual conflict (non-zero) -> ``git rebase --abort`` (leave worktree
+           clean) -> ``(False, "rebase conflict: <files>")`` (AC-3).
+      3. clean rebase -> ``git push --force-with-lease origin <branch>`` — ONLY the
+         task branch, NEVER ``main`` (AC-7) -> ``(True, "rebased onto origin/main")``.
+
+    Never-raise (AC-9): any git/OS error -> ``(False, "<reason>")``.
+    """
+    try:
+        wt = ensure_worktree(repo, branch)
+    except Exception as e:  # noqa: BLE001 - never-raise contract
+        return False, f"rebase setup error: {e}"
+
+    try:
+        subprocess.run(
+            ["git", "-C", wt, "fetch", "origin", "main"],
+            capture_output=True, timeout=_FETCH_TIMEOUT,
+        )
+        r = subprocess.run(
+            ["git", "-C", wt, "rebase", "origin/main"],
+            capture_output=True, text=True, timeout=_REBASE_TIMEOUT,
+        )
+        if r.returncode != 0:
+            files = _conflicted_files(wt)
+            subprocess.run(
+                ["git", "-C", wt, "rebase", "--abort"],
+                capture_output=True, timeout=_SHORT_TIMEOUT,
+            )
+            logger.warning("auto_rebase: conflict on %s/%s: %s", repo, branch, files)
+            return False, f"rebase conflict: {files}"
+
+        # Clean rebase -> push ONLY the task branch with a lease (never main).
+        p = subprocess.run(
+            ["git", "-C", wt, "push", "--force-with-lease", "origin", branch],
+            capture_output=True, text=True, timeout=_PUSH_TIMEOUT,
+        )
+        if p.returncode != 0:
+            detail = (p.stderr or p.stdout or "").strip()[:200]
+            logger.warning("auto_rebase: push failed on %s/%s: %s", repo, branch, detail)
+            return False, f"push --force-with-lease failed: {detail}"
+
+        logger.info("auto_rebase: %s/%s rebased onto origin/main and pushed", repo, branch)
+        return True, "rebased onto origin/main"
+    except subprocess.TimeoutExpired:
+        # Leave no half-finished rebase behind.
+        try:
+            subprocess.run(
+                ["git", "-C", wt, "rebase", "--abort"],
+                capture_output=True, timeout=_SHORT_TIMEOUT,
+            )
+        except (subprocess.SubprocessError, OSError):
+            pass
+        return False, "rebase timeout"
+    except (subprocess.SubprocessError, OSError) as e:
+        return False, f"rebase error: {e}"
+
+
+# ---------------------------------------------------------------------------
+# re-test in the caught-up worktree
+# ---------------------------------------------------------------------------
+def retest_branch(repo: str, branch: str) -> tuple[bool, str]:
+    """Run the project test-suite in the (already caught-up) branch worktree.
+
+    Command: ``python -m pytest <merge_retest_target>`` (default ``tests/``),
+    matching the orchestrator CI / check_tests_local pattern. Bounded by
+    ``settings.merge_retest_timeout_s``.
+
+    Returns:
+      * ``(True,  "re-test green")``                 — pytest rc == 0
+      * ``(False, "re-test timeout after <T>s")``    — exceeded the timeout (AC-6)
+      * ``(False, "re-test failed: ...<tail>")``     — non-zero rc, with output tail
+    Never-raise (AC-9): any setup/OS error -> ``(False, "<reason>")``.
+    """
+    wt = get_worktree_path(repo, branch)
+    if not os.path.isdir(wt):
+        # Caller usually rebased first (worktree exists); ensure as a fallback.
+        try:
+            wt = ensure_worktree(repo, branch)
+        except Exception as e:  # noqa: BLE001 - never-raise contract
+            return False, f"re-test setup error: {e}"
+
+    target = settings.merge_retest_target or "tests/"
+    timeout = settings.merge_retest_timeout_s
+    try:
+        r = subprocess.run(
+            ["python", "-m", "pytest", target, "-q"],
+            cwd=wt, capture_output=True, text=True, timeout=timeout,
+        )
+    except subprocess.TimeoutExpired:
+        logger.warning("retest_branch: timeout (%ss) on %s/%s", timeout, repo, branch)
+        return False, f"re-test timeout after {timeout}s"
+    except (subprocess.SubprocessError, OSError) as e:
+        return False, f"re-test error: {e}"
+
+    if r.returncode == 0:
+        return True, "re-test green"
+    tail = ((r.stdout or "") + (r.stderr or ""))[-500:]
+    logger.warning("retest_branch: red on %s/%s", repo, branch)
+    return False, f"re-test failed: ...{tail}"
+
+
+# ---------------------------------------------------------------------------
+# merge-lease (serialise catch-up + re-test + merge per repo)
+# ---------------------------------------------------------------------------
+def _lease_path(repo: str) -> str:
+    """Filesystem path of the per-repo merge lease (no schema change, ТЗ §4)."""
+    return os.path.join(settings.repos_dir, f".merge-lease-{repo}.json")
+
+
+def _read_lease(path: str) -> dict | None:
+    """Read+parse the lease file; None if missing or corrupt (never-raise)."""
+    try:
+        with open(path, "r", encoding="utf-8") as f:
+            return json.loads(f.read())
+    except FileNotFoundError:
+        return None
+    except (OSError, ValueError) as e:
+        logger.warning("merge-lease read error at %s: %s", path, e)
+        return None
+
+
+def _write_lease(path: str, holder: dict) -> None:
+    """Atomically (O_CREAT|O_EXCL) write the lease; raises FileExistsError if held."""
+    fd = os.open(path, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o644)
+    try:
+        os.write(fd, json.dumps(holder).encode("utf-8"))
+    finally:
+        os.close(fd)
+
+
+def acquire_merge_lease(
+    repo: str, branch: str, work_item_id: str | None = None, task_id: int | None = None
+) -> tuple[bool, str]:
+    """Try to acquire the per-repo merge lease. **Non-blocking** (anti-deadlock).
+
+    Holder identity is the task ``branch`` (stable, one branch per task). Outcomes:
+      * no lease file        -> acquire, write metadata -> ``(True, "lease acquired")``
+      * lease held by self   -> idempotent re-acquire (restart/retry) -> ``(True, "lease already held")``
+      * lease held by other, age <  merge_lock_timeout_s -> ``(False, "merge-lock busy")``
+      * lease held by other, age >= merge_lock_timeout_s -> stale -> reclaim with a
+        ``logger.warning`` (the holder process died without releasing) -> ``(True, ...)``
+
+    Never-raise: any unexpected error -> ``(False, "merge-lock busy")`` so the caller
+    DEFERS and retries rather than burning a developer retry on an infra hiccup.
+    """
+    path = _lease_path(repo)
+    holder = {
+        "branch": branch,
+        "work_item_id": work_item_id,
+        "task_id": task_id,
+        "acquired_at": time.time(),
+        "pid": os.getpid(),
+    }
+    try:
+        try:
+            _write_lease(path, holder)
+            logger.info("merge-lease acquired for %s by %s", repo, branch)
+            return True, "lease acquired"
+        except FileExistsError:
+            pass
+
+        existing = _read_lease(path)
+        if existing is None:
+            # Corrupt/empty lease file — reclaim it.
+            _force_write_lease(path, holder)
+            logger.warning("merge-lease for %s was corrupt; reclaimed by %s", repo, branch)
+            return True, "lease reclaimed (corrupt)"
+
+        if existing.get("branch") == branch:
+            return True, "lease already held"
+
+        age = time.time() - float(existing.get("acquired_at") or 0)
+        if age >= settings.merge_lock_timeout_s:
+            _force_write_lease(path, holder)
+            logger.warning(
+                "merge-lease for %s was stale (age %.0fs >= %ss, holder=%s); reclaimed by %s",
+                repo, age, settings.merge_lock_timeout_s, existing.get("branch"), branch,
+            )
+            return True, "lease reclaimed (stale)"
+
+        logger.info(
+            "merge-lease for %s busy (held by %s, age %.0fs); %s defers",
+            repo, existing.get("branch"), age, branch,
+        )
+        return False, "merge-lock busy"
+    except Exception as e:  # noqa: BLE001 - never-raise contract
+        logger.warning("acquire_merge_lease unexpected error for %s/%s: %s", repo, branch, e)
+        return False, "merge-lock busy"
+
+
+def _force_write_lease(path: str, holder: dict) -> None:
+    """Overwrite the lease (used for stale/corrupt reclaim). Best-effort."""
+    try:
+        with open(path, "w", encoding="utf-8") as f:
+            f.write(json.dumps(holder))
+    except OSError as e:
+        logger.warning("merge-lease force-write error at %s: %s", path, e)
+
+
+def release_merge_lease(repo: str, branch: str | None = None) -> None:
+    """Release the per-repo merge lease. **Idempotent** and **holder-aware**.
+
+    If ``branch`` is given, the lease is removed ONLY when the current holder's
+    branch matches (so a delayed release from an already-merged task can never
+    delete a lease a DIFFERENT task acquired afterwards). With ``branch=None`` the
+    release is unconditional (best-effort backstop). Never raises.
+    """
+    path = _lease_path(repo)
+    try:
+        if branch is not None:
+            existing = _read_lease(path)
+            if existing is not None and existing.get("branch") != branch:
+                logger.info(
+                    "merge-lease release skipped for %s: holder=%s != %s",
+                    repo, existing.get("branch"), branch,
+                )
+                return
+        os.remove(path)
+        logger.info("merge-lease released for %s (%s)", repo, branch or "force")
+    except FileNotFoundError:
+        return
+    except OSError as e:
+        logger.warning("merge-lease release error for %s: %s", repo, e)

src/qg/checks.py

@@ -621,6 +621,87 @@ def check_staging_status(repo: str, work_item_id: str, branch: str | None = None
     return False, "Staging log not found (15-staging-log.md)"
 
 
+def _merge_gate_applies(repo: str) -> bool:
+    """Whether the merge-gate is REAL for this repo (ORCH-043, conditional rollout).
+
+    Mirrors the ORCH-35 conditional staging-gate. ``merge_gate_repos`` is a CSV of
+    repos where the gate is enforced; when empty the gate is real ONLY for the
+    self-hosting repo (``orchestrator``). Other repos -> conditional no-op.
+    """
+    raw = (settings.merge_gate_repos or "").strip()
+    if raw:
+        allowed = {r.strip().lower() for r in raw.split(",") if r.strip()}
+        return (repo or "").strip().lower() in allowed
+    return is_self_hosting_repo(repo)
+
+
+def check_branch_mergeable(repo: str, work_item_id: str, branch: str) -> tuple[bool, str]:
+    """ORCH-043 merge-gate: validate the branch against the CURRENT origin/main
+    immediately before the deployer merges its PR (deploy-staging -> deploy edge).
+
+    Deterministic, no LLM. Algorithm (ADR-001 §4):
+      1. Conditionality: merge_gate_enabled=False -> (True, "merge-gate disabled");
+         repo where the gate is not real -> (True, "merge-gate N/A for <repo>").
+      2. Acquire the per-repo merge lease (NON-blocking). Busy -> (False, "merge-lock
+         busy") — a SIGNAL for the engine to DEFER (not a code fault, no rollback).
+      3. Double-check "behind origin/main" UNDER the lease (main may have moved while
+         we waited). Not behind -> (True, "branch up-to-date with main"); lease HELD.
+      4. Behind -> auto_rebase_onto_main:
+           - conflict           -> release lease -> (False, "rebase conflict: ...")
+           - clean -> retest_branch:
+               green            -> (True, "rebased onto main, re-test green"); lease HELD
+               red/timeout      -> release lease -> (False, "re-test ... after rebase")
+      5. On SUCCESS the lease is HELD until the actual merge (released on PR-merged
+         webhook / deploy->done / rollback). On any FAILURE the lease is released.
+
+    Never-raise (AC-9): any internal error -> (False, "<reason>") with the lease
+    released; an exception never escapes into advance_stage.
+    """
+    # Imported lazily so qg.checks stays importable without the merge_gate deps in
+    # minimal/test contexts and to avoid an import cycle surprise.
+    from .. import merge_gate
+
+    try:
+        if not settings.merge_gate_enabled:
+            return True, "merge-gate disabled"
+        if not _merge_gate_applies(repo):
+            return True, f"merge-gate N/A for {repo}"
+
+        acquired, reason = merge_gate.acquire_merge_lease(repo, branch, work_item_id)
+        if not acquired:
+            # "merge-lock busy" -> caller defers; lease NOT held by us, nothing to release.
+            return False, reason
+
+        try:
+            # Double-check under the lease: another task may have just merged.
+            if not merge_gate.branch_is_behind_main(repo, branch):
+                logger.info("check_branch_mergeable: %s up-to-date with main", branch)
+                return True, "branch up-to-date with main"
+
+            ok, rb_reason = merge_gate.auto_rebase_onto_main(repo, branch)
+            if not ok:
+                merge_gate.release_merge_lease(repo, branch)
+                return False, rb_reason  # "rebase conflict: ..."
+
+            ok_t, t_reason = merge_gate.retest_branch(repo, branch)
+            if ok_t:
+                logger.info("check_branch_mergeable: %s rebased + re-test green", branch)
+                return True, "rebased onto main, re-test green"
+
+            merge_gate.release_merge_lease(repo, branch)
+            if "timeout" in t_reason:
+                return False, t_reason  # "re-test timeout after <T>s" (AC-6)
+            tail = t_reason.removeprefix("re-test failed: ")
+            return False, f"re-test failed after rebase: {tail}"
+        except Exception as e:  # noqa: BLE001 - never-raise; always release on error
+            merge_gate.release_merge_lease(repo, branch)
+            logger.error("check_branch_mergeable inner error for %s/%s: %s", repo, branch, e)
+            return False, f"merge-gate error: {e}"
+    except Exception as e:  # noqa: BLE001 - outer never-raise guard
+        logger.error("check_branch_mergeable error for %s/%s: %s", repo, branch, e)
+        return False, f"merge-gate error: {e}"
+
+
 # Registry for dynamic lookup by name
 QG_CHECKS = {
     "check_analysis_approved": check_analysis_approved,
@@ -633,4 +714,5 @@ QG_CHECKS = {
     "check_tests_local": check_tests_local,
     "check_deploy_status": check_deploy_status,
     "check_staging_status": check_staging_status,
+    "check_branch_mergeable": check_branch_mergeable,
 }

src/stage_engine.py

@@ -34,6 +34,7 @@ from .stages import get_next_stage, get_qg_for_stage, get_agent_for_stage
 from .git_worktree import get_worktree_path
 from .review_parse import extract_review_findings, extract_test_failures
 from .qg.checks import QG_CHECKS
+from . import merge_gate
 from .notifications import (
     notify_stage_change,
     notify_qg_failure,
@@ -239,6 +240,18 @@ def advance_stage(
             result.note = f"qg '{qg_name}' not in registry"
             return result
 
+        # --- ORCH-043 merge-gate sub-gate (deploy-staging -> deploy edge) -----
+        # AFTER check_staging_status passed and BEFORE we advance to `deploy` /
+        # launch the deployer that merges the PR. Not a STAGE_TRANSITIONS entry —
+        # it is an edge sub-gate triggered by the same "staging-deployer finished"
+        # event. If it intervenes (defer on busy-lock, or rollback on conflict /
+        # red re-test) it owns the outcome and we return without advancing.
+        if current_stage == "deploy-staging":
+            if _handle_merge_gate(
+                task_id, current_stage, repo, work_item_id, branch, agent, result
+            ):
+                return result
+
         # --- Advance ---------------------------------------------------------
         update_task_stage(task_id, next_stage)
         # Telegram live tracker: the analysis->architecture advance is the human
@@ -274,6 +287,15 @@ def advance_stage(
             except Exception as e:
                 logger.error(f"Task {task_id}: failed to set Plane Done: {e}")
 
+        # ORCH-043: the merge has landed (deploy->done). Release the merge lease as
+        # a backstop in case the PR-merged webhook was lost (holder-aware no-op if a
+        # different task already owns it). Never raises.
+        if next_stage == "done":
+            try:
+                merge_gate.release_merge_lease(repo, branch)
+            except Exception as e:  # noqa: BLE001 - defensive
+                logger.warning(f"Task {task_id}: merge-lease release on done failed: {e}")
+
         # --- Launch the next agent (ORCH-4 fix: current_stage, not next) -----
         next_agent = get_agent_for_stage(current_stage)
         if next_agent:
@@ -565,6 +587,12 @@ def _handle_qg_failure_rollbacks(
         notify_stage_change(task_id, current_stage, "development")
         plane_notify_stage(work_item_id, current_stage, "development")
         result.rolled_back_to = "development"
+        # ORCH-043: deploy failed -> no merge will complete; release the lease so the
+        # next task isn't blocked until the lease ages out (holder-aware no-op).
+        try:
+            merge_gate.release_merge_lease(repo, branch)
+        except Exception as e:  # noqa: BLE001 - defensive
+            logger.warning(f"Task {task_id}: merge-lease release on deploy-fail failed: {e}")
         set_issue_blocked(work_item_id)
         notify_qg_failure(task_id, "deploy", "check_deploy_status", reason)
         plane_add_comment(
@@ -582,3 +610,155 @@ def _handle_qg_failure_rollbacks(
             f"Task {task_id}: deployer verdict FAILED, rolled back deploy -> "
             f"development ({reason})"
         )
+
+
+# ---------------------------------------------------------------------------
+# ORCH-043: merge-gate sub-gate on the deploy-staging -> deploy edge
+# ---------------------------------------------------------------------------
+def _merge_defer_count(task_id: int) -> int:
+    """How many times this task has already been deferred by the merge-gate.
+
+    Counted from the persisted jobs queue (restart-safe) by the defer marker in
+    task_content, so a service restart never resets the defer budget.
+    """
+    conn = get_db()
+    n = conn.execute(
+        "SELECT COUNT(*) FROM jobs WHERE task_id=? AND task_content LIKE '%merge-gate defer%'",
+        (task_id,),
+    ).fetchone()[0]
+    conn.close()
+    return n
+
+
+def _handle_merge_gate(
+    task_id, current_stage, repo, work_item_id, branch, agent, result: AdvanceResult
+) -> bool:
+    """Run check_branch_mergeable on the deploy-staging -> deploy edge.
+
+    Returns True if the gate INTERVENED (the caller must return without advancing):
+      * "merge-lock busy"            -> DEFER (re-queue the staging-deployer with a
+                                        delay; the task stays on deploy-staging). Code
+                                        is fine, so NO rollback and no developer retry.
+      * conflict / red re-test       -> ROLLBACK to development (+ developer retry,
+                                        capped by MAX_DEVELOPER_RETRIES).
+    Returns False when the gate PASSED (branch up-to-date, or rebased + re-test green)
+    so advance_stage proceeds to `deploy` and launches the deployer that merges. On a
+    PASS the merge lease is HELD until the actual merge (released on PR-merged webhook
+    / deploy->done / rollback).
+    """
+    passed, reason = _run_qg("check_branch_mergeable", repo, work_item_id, branch)
+    if passed:
+        logger.info(f"Task {task_id}: merge-gate passed ({reason})")
+        return False
+
+    result.qg_name = "check_branch_mergeable"
+    result.qg_passed = False
+    result.qg_reason = reason
+
+    if reason == "merge-lock busy":
+        _handle_merge_gate_defer(
+            task_id, current_stage, repo, work_item_id, branch, result
+        )
+        return True
+
+    _handle_merge_gate_rollback(
+        task_id, current_stage, repo, work_item_id, branch, reason, result
+    )
+    return True
+
+
+def _handle_merge_gate_defer(
+    task_id, current_stage, repo, work_item_id, branch, result: AdvanceResult
+):
+    """merge-lock busy -> DEFER: re-queue the staging-deployer after a delay.
+
+    Non-blocking: the worker slot is freed (anti-deadlock at max_concurrency=1) so
+    the lease HOLDER can finish merging. The task remains on deploy-staging; a later
+    staging-deployer run re-evaluates the gate. Bounded by merge_defer_max_attempts.
+    """
+    defers = _merge_defer_count(task_id)
+    if defers < settings.merge_defer_max_attempts:
+        task_desc = (
+            f"Work item: {work_item_id}\nRepo: {repo}\nBranch: {branch}\n"
+            f"Stage: deploy-staging\nNote: merge-gate defer "
+            f"(attempt {defers + 1}/{settings.merge_defer_max_attempts}) — "
+            f"merge-lock busy, retrying after {settings.merge_defer_delay_s}s."
+        )
+        new_job = enqueue_job(
+            "deployer", repo, task_desc, task_id=task_id,
+            available_at_delay_s=settings.merge_defer_delay_s,
+        )
+        result.enqueued_agent = "deployer"
+        result.enqueued_job_id = new_job
+        result.note = "merge-gate-deferred"
+        logger.info(
+            f"Task {task_id}: merge-lock busy, deferred deployer "
+            f"(job_id={new_job}, attempt {defers + 1}/{settings.merge_defer_max_attempts})"
+        )
+    else:
+        set_issue_blocked(work_item_id)
+        send_telegram(
+            f"\U0001f6a8 {work_item_id}: merge-gate defer limit "
+            f"({settings.merge_defer_max_attempts}) reached (merge-lock busy). "
+            f"Manual intervention needed."
+        )
+        result.alerted = True
+        result.note = "merge-gate-defer-exhausted"
+        logger.error(
+            f"Task {task_id}: merge-gate defer attempts exhausted "
+            f"({settings.merge_defer_max_attempts})"
+        )
+
+
+def _handle_merge_gate_rollback(
+    task_id, current_stage, repo, work_item_id, branch, reason, result: AdvanceResult
+):
+    """Rebase conflict / red re-test -> ROLLBACK to development + developer retry.
+
+    Mirrors the staging/deploy rollback pattern but is capped by
+    MAX_DEVELOPER_RETRIES (AC-11 / TC-22: no infinite bounce). The merge lease was
+    already released by check_branch_mergeable on failure; a defensive holder-aware
+    release here is a harmless no-op.
+    """
+    update_task_stage(task_id, "development")
+    notify_stage_change(task_id, current_stage, "development")
+    plane_notify_stage(work_item_id, current_stage, "development")
+    result.rolled_back_to = "development"
+    set_issue_in_progress(work_item_id)
+    try:
+        merge_gate.release_merge_lease(repo, branch)
+    except Exception as e:  # noqa: BLE001 - defensive
+        logger.warning(f"Task {task_id}: merge-lease release on rollback failed: {e}")
+    notify_qg_failure(task_id, current_stage, "check_branch_mergeable", reason)
+    plane_add_comment(
+        work_item_id,
+        f"❌ Merge-gate FAILED ({reason}). Rolled back to development. "
+        f"Developer нужен для фикса.",
+        author="deployer",
+    )
+    retry_count = _developer_retry_count(task_id)
+    if retry_count < MAX_DEVELOPER_RETRIES:
+        task_desc = (
+            f"Work item: {work_item_id}\nRepo: {repo}\nBranch: {branch}\n"
+            f"Stage: development\nNote: Merge-gate failed "
+            f"(attempt {retry_count + 1}/{MAX_DEVELOPER_RETRIES}). "
+            f"Причина: {reason}."
+        )
+        new_job = enqueue_job("developer", repo, task_desc, task_id=task_id)
+        result.enqueued_agent = "developer"
+        result.enqueued_job_id = new_job
+        logger.info(
+            f"Task {task_id}: merge-gate FAILED, enqueued developer (job_id={new_job})"
+        )
+    else:
+        set_issue_blocked(work_item_id)
+        send_telegram(
+            f"\U0001f6a8 {work_item_id}: Merge-gate still failing after "
+            f"{MAX_DEVELOPER_RETRIES} developer retries ({reason}). "
+            f"Manual intervention needed."
+        )
+        result.alerted = True
+    logger.error(
+        f"Task {task_id}: merge-gate FAILED, rolled back deploy-staging -> "
+        f"development ({reason})"
+    )

src/webhooks/gitea.py

@@ -334,6 +334,15 @@ async def handle_pr(payload: dict):
                 logger.error(f"Task {task_id}: max retries reached, needs manual intervention")
 
     elif action == "closed" and pr.get("merged", False):
+        # ORCH-043: the branch's PR just merged into main -> release the per-repo
+        # merge lease this task held from the merge-gate (holder-aware by branch, so
+        # it can't clobber a lease another task acquired afterwards). Never raises.
+        try:
+            from ..merge_gate import release_merge_lease
+            release_merge_lease(repo_name, head_branch)
+        except Exception as e:  # noqa: BLE001 - defensive, never block the webhook
+            logger.warning(f"Task {task_id}: merge-lease release on PR-merge failed: {e}")
+
         # BUG 8 (second door): at the deploy stage `done` is gated by the
         # deployer's verdict (check_deploy_status via advance_stage), NOT by the
         # fact that the PR was merged. The deployer merges the PR at the START of

tests/test_config.py

@@ -25,3 +25,50 @@ def test_tracker_mode_reads_env_arbitrary(monkeypatch):
     # -> edit) happens in notifications, not here (AC-1/AC-2 split).
     monkeypatch.setenv("ORCH_TRACKER_MODE", "garbage")
     assert Settings().tracker_mode == "garbage"
+
+
+# ---------------------------------------------------------------------------
+# ORCH-043 / TC-25: merge-gate settings defaults + env override.
+# ---------------------------------------------------------------------------
+_MERGE_ENV = (
+    "ORCH_MERGE_GATE_ENABLED",
+    "ORCH_MERGE_GATE_REPOS",
+    "ORCH_MERGE_RETEST_TIMEOUT_S",
+    "ORCH_MERGE_RETEST_TARGET",
+    "ORCH_MERGE_LOCK_TIMEOUT_S",
+    "ORCH_MERGE_DEFER_DELAY_S",
+    "ORCH_MERGE_DEFER_MAX_ATTEMPTS",
+)
+
+
+def test_merge_gate_settings_defaults(monkeypatch):
+    """TC-25 / AC-10: documented defaults when no env is set."""
+    for name in _MERGE_ENV:
+        monkeypatch.delenv(name, raising=False)
+    s = Settings()
+    assert s.merge_gate_enabled is True
+    assert s.merge_gate_repos == ""
+    assert s.merge_retest_timeout_s == 600
+    assert s.merge_retest_target == "tests/"
+    assert s.merge_lock_timeout_s == 300
+    assert s.merge_defer_delay_s == 60
+    assert s.merge_defer_max_attempts == 5
+
+
+def test_merge_gate_settings_env_override(monkeypatch):
+    """TC-25 / AC-10: each field is read from its ORCH_* env var."""
+    monkeypatch.setenv("ORCH_MERGE_GATE_ENABLED", "false")
+    monkeypatch.setenv("ORCH_MERGE_GATE_REPOS", "orchestrator,enduro-trails")
+    monkeypatch.setenv("ORCH_MERGE_RETEST_TIMEOUT_S", "120")
+    monkeypatch.setenv("ORCH_MERGE_RETEST_TARGET", "tests/unit")
+    monkeypatch.setenv("ORCH_MERGE_LOCK_TIMEOUT_S", "90")
+    monkeypatch.setenv("ORCH_MERGE_DEFER_DELAY_S", "5")
+    monkeypatch.setenv("ORCH_MERGE_DEFER_MAX_ATTEMPTS", "9")
+    s = Settings()
+    assert s.merge_gate_enabled is False
+    assert s.merge_gate_repos == "orchestrator,enduro-trails"
+    assert s.merge_retest_timeout_s == 120
+    assert s.merge_retest_target == "tests/unit"
+    assert s.merge_lock_timeout_s == 90
+    assert s.merge_defer_delay_s == 5
+    assert s.merge_defer_max_attempts == 9

tests/test_merge_gate.py

@@ -0,0 +1,301 @@
+"""ORCH-043: tests for src/merge_gate core (TC-01..TC-11).
+
+Git tests use REAL local repos in tmp (a bare 'origin' + a main clone), so
+fetch / merge-base / rebase / push --force-with-lease are exercised without
+network, mirroring tests/test_git_worktree.py. The re-test (pytest) and lease
+units are isolated with monkeypatch / tmp files.
+"""
+import json
+import os
+import subprocess
+import tempfile
+import time
+
+import pytest
+
+# Env before importing app modules (same convention as the other suites).
+_test_db = os.path.join(tempfile.gettempdir(), "test_orchestrator_merge_gate.db")
+os.environ["ORCH_DB_PATH"] = _test_db
+os.environ["ORCH_REPOS_DIR"] = tempfile.gettempdir()
+os.environ["ORCH_GITEA_TOKEN"] = "test-token"
+os.environ["ORCH_PLANE_API_TOKEN"] = "test-token"
+
+from src import git_worktree, merge_gate  # noqa: E402
+
+
+def _git(cwd, *args):
+    return subprocess.run(["git", "-C", cwd, *args], capture_output=True, text=True)
+
+
+def _origin_sha(origin, ref):
+    return _git(str(origin), "rev-parse", ref).stdout.strip()
+
+
+@pytest.fixture
+def repos(tmp_path, monkeypatch):
+    """Bare 'origin' (main@C1) + main clone + two feature branches branched from C0.
+
+    Layout:
+      C0  README.md
+      feature/behind   : C0 + adds f.txt           (rebases cleanly onto C1)
+      feature/conflict : C0 + edits README.md       (textual conflict with C1)
+      feature/uptodate : branched from C1            (already contains origin/main)
+      main C1          : edits README.md + adds other.txt
+    Returns (repo_name, origin_path).
+    """
+    repo = "orchestrator"
+    repos_dir = tmp_path / "repos"
+    wt_dir = tmp_path / "repos" / "_wt"
+    repos_dir.mkdir(parents=True)
+
+    monkeypatch.setattr(merge_gate.settings, "repos_dir", str(repos_dir))
+    monkeypatch.setattr(git_worktree.settings, "repos_dir", str(repos_dir))
+    monkeypatch.setattr(git_worktree.settings, "worktrees_dir", str(wt_dir))
+
+    origin = tmp_path / "origin.git"
+    subprocess.run(["git", "init", "--bare", "-b", "main", str(origin)], capture_output=True)
+
+    seed = tmp_path / "seed"
+    seed.mkdir()
+    _git(str(seed), "init", "-b", "main")
+    _git(str(seed), "config", "user.email", "t@t")
+    _git(str(seed), "config", "user.name", "t")
+    (seed / "README.md").write_text("base\n")
+    _git(str(seed), "add", ".")
+    _git(str(seed), "commit", "-m", "C0")
+    _git(str(seed), "remote", "add", "origin", str(origin))
+    _git(str(seed), "push", "origin", "main")
+
+    # Branches off C0.
+    _git(str(seed), "checkout", "-b", "feature/behind")
+    (seed / "f.txt").write_text("feature\n")
+    _git(str(seed), "add", ".")
+    _git(str(seed), "commit", "-m", "feat: add f.txt")
+    _git(str(seed), "push", "origin", "feature/behind")
+
+    _git(str(seed), "checkout", "main")
+    _git(str(seed), "checkout", "-b", "feature/conflict")
+    (seed / "README.md").write_text("feature readme\n")
+    _git(str(seed), "add", ".")
+    _git(str(seed), "commit", "-m", "feat: edit README")
+    _git(str(seed), "push", "origin", "feature/conflict")
+
+    # Advance main to C1.
+    _git(str(seed), "checkout", "main")
+    (seed / "README.md").write_text("main v2\n")
+    (seed / "other.txt").write_text("main change\n")
+    _git(str(seed), "add", ".")
+    _git(str(seed), "commit", "-m", "C1")
+    _git(str(seed), "push", "origin", "main")
+
+    # Branch that already contains C1.
+    _git(str(seed), "checkout", "-b", "feature/uptodate")
+    (seed / "g.txt").write_text("uptodate\n")
+    _git(str(seed), "add", ".")
+    _git(str(seed), "commit", "-m", "feat: on top of C1")
+    _git(str(seed), "push", "origin", "feature/uptodate")
+
+    # Main clone at repos_dir/<repo>.
+    main_clone = repos_dir / repo
+    subprocess.run(["git", "clone", str(origin), str(main_clone)], capture_output=True)
+    _git(str(main_clone), "config", "user.email", "t@t")
+    _git(str(main_clone), "config", "user.name", "t")
+    return repo, origin
+
+
+# ---------------------------------------------------------------------------
+# TC-01..03: branch_is_behind_main
+# ---------------------------------------------------------------------------
+def test_tc01_behind_when_main_ahead(repos):
+    repo, _ = repos
+    assert merge_gate.branch_is_behind_main(repo, "feature/behind") is True
+
+
+def test_tc02_not_behind_when_branch_contains_main(repos):
+    repo, _ = repos
+    assert merge_gate.branch_is_behind_main(repo, "feature/uptodate") is False
+
+
+def test_tc03_never_raises_on_bad_repo(monkeypatch, tmp_path):
+    # Point repos_dir at an empty dir -> ensure_worktree raises -> caught -> True.
+    monkeypatch.setattr(merge_gate.settings, "repos_dir", str(tmp_path / "nope"))
+    monkeypatch.setattr(git_worktree.settings, "repos_dir", str(tmp_path / "nope"))
+    monkeypatch.setattr(git_worktree.settings, "worktrees_dir", str(tmp_path / "_wt"))
+    result = merge_gate.branch_is_behind_main("orchestrator", "feature/x")
+    assert result is True  # safe bool, not an exception
+
+
+# ---------------------------------------------------------------------------
+# TC-04..06: auto_rebase_onto_main
+# ---------------------------------------------------------------------------
+def test_tc04_clean_catchup_pushes_with_lease(repos):
+    repo, origin = repos
+    main_before = _origin_sha(origin, "main")
+
+    ok, reason = merge_gate.auto_rebase_onto_main(repo, "feature/behind")
+    assert ok is True, reason
+
+    # origin/main must be UNTOUCHED (AC-7).
+    assert _origin_sha(origin, "main") == main_before
+    # The pushed branch now contains origin/main (origin/main is its ancestor).
+    rc = subprocess.run(
+        ["git", "-C", str(origin), "merge-base", "--is-ancestor",
+         "main", "feature/behind"],
+        capture_output=True,
+    ).returncode
+    assert rc == 0
+    # And it carries main's new file plus its own.
+    assert _git(str(origin), "cat-file", "-e", "feature/behind:other.txt").returncode == 0
+    assert _git(str(origin), "cat-file", "-e", "feature/behind:f.txt").returncode == 0
+
+
+def test_tc05_conflict_aborts_clean_and_reports(repos):
+    repo, origin = repos
+    main_before = _origin_sha(origin, "main")
+    branch_before = _origin_sha(origin, "feature/conflict")
+
+    ok, reason = merge_gate.auto_rebase_onto_main(repo, "feature/conflict")
+    assert ok is False
+    assert "rebase conflict" in reason
+    # main untouched, branch NOT force-pushed past the conflict.
+    assert _origin_sha(origin, "main") == main_before
+    assert _origin_sha(origin, "feature/conflict") == branch_before
+    # Worktree left clean (no rebase in progress).
+    wt = git_worktree.get_worktree_path(repo, "feature/conflict")
+    assert not os.path.isdir(os.path.join(wt, ".git", "rebase-merge"))
+    assert not os.path.isdir(os.path.join(wt, ".git", "rebase-apply"))
+
+
+def test_tc06_never_pushes_main(repos, monkeypatch):
+    repo, origin = repos
+    calls = []
+    real_run = subprocess.run
+
+    def _spy(cmd, *a, **k):
+        if isinstance(cmd, list):
+            calls.append(cmd)
+        return real_run(cmd, *a, **k)
+
+    monkeypatch.setattr(merge_gate.subprocess, "run", _spy)
+    merge_gate.auto_rebase_onto_main(repo, "feature/behind")
+
+    pushes = [c for c in calls if "push" in c]
+    assert pushes, "expected at least one push"
+    for c in pushes:
+        # Never push main; force only via --force-with-lease on the task branch.
+        assert "main" not in c, f"push touched main: {c}"
+        assert "--force-with-lease" in c
+        assert "feature/behind" in c
+        # Hard force must never be used.
+        assert "--force" not in c or "--force-with-lease" in c
+        assert "-f" not in c
+
+
+# ---------------------------------------------------------------------------
+# TC-07..09: retest_branch (isolated from real pytest)
+# ---------------------------------------------------------------------------
+@pytest.fixture
+def fake_worktree(tmp_path, monkeypatch):
+    wt = tmp_path / "wt"
+    wt.mkdir()
+    monkeypatch.setattr(merge_gate, "get_worktree_path", lambda repo, branch: str(wt))
+    return str(wt)
+
+
+def test_tc07_retest_green(fake_worktree, monkeypatch):
+    monkeypatch.setattr(
+        merge_gate.subprocess, "run",
+        lambda *a, **k: subprocess.CompletedProcess(a, 0, "1 passed", ""),
+    )
+    ok, reason = merge_gate.retest_branch("orchestrator", "feature/x")
+    assert ok is True
+    assert reason == "re-test green"
+
+
+def test_tc08_retest_red_with_tail(fake_worktree, monkeypatch):
+    monkeypatch.setattr(
+        merge_gate.subprocess, "run",
+        lambda *a, **k: subprocess.CompletedProcess(
+            a, 1, "FAILED tests/test_x.py::t - AssertionError\n1 failed", ""
+        ),
+    )
+    ok, reason = merge_gate.retest_branch("orchestrator", "feature/x")
+    assert ok is False
+    assert "re-test failed" in reason
+    assert "AssertionError" in reason  # output tail embedded
+
+
+def test_tc09_retest_timeout(fake_worktree, monkeypatch):
+    def _boom(*a, **k):
+        raise subprocess.TimeoutExpired(cmd="pytest", timeout=1)
+
+    monkeypatch.setattr(merge_gate.settings, "merge_retest_timeout_s", 1)
+    monkeypatch.setattr(merge_gate.subprocess, "run", _boom)
+    ok, reason = merge_gate.retest_branch("orchestrator", "feature/x")
+    assert ok is False
+    assert "re-test timeout" in reason
+
+
+# ---------------------------------------------------------------------------
+# TC-10..11: merge-lease (serialisation)
+# ---------------------------------------------------------------------------
+@pytest.fixture
+def lease_dir(tmp_path, monkeypatch):
+    d = tmp_path / "repos"
+    d.mkdir()
+    monkeypatch.setattr(merge_gate.settings, "repos_dir", str(d))
+    monkeypatch.setattr(merge_gate.settings, "merge_lock_timeout_s", 300)
+    return d
+
+
+def test_tc10_second_acquire_busy_until_released(lease_dir):
+    repo = "orchestrator"
+    ok, _ = merge_gate.acquire_merge_lease(repo, "feature/A", "ORCH-1")
+    assert ok is True
+
+    # A different branch cannot acquire while held.
+    ok2, reason2 = merge_gate.acquire_merge_lease(repo, "feature/B", "ORCH-2")
+    assert ok2 is False
+    assert reason2 == "merge-lock busy"
+
+    # Same holder is idempotent.
+    ok_self, _ = merge_gate.acquire_merge_lease(repo, "feature/A", "ORCH-1")
+    assert ok_self is True
+
+    # Release (holder-aware) frees it for B.
+    merge_gate.release_merge_lease(repo, "feature/A")
+    ok3, _ = merge_gate.acquire_merge_lease(repo, "feature/B", "ORCH-2")
+    assert ok3 is True
+
+
+def test_tc10_release_is_holder_aware(lease_dir):
+    repo = "orchestrator"
+    merge_gate.acquire_merge_lease(repo, "feature/A", "ORCH-1")
+    # A stale release from a DIFFERENT branch must NOT drop A's lease.
+    merge_gate.release_merge_lease(repo, "feature/OTHER")
+    ok, reason = merge_gate.acquire_merge_lease(repo, "feature/B", "ORCH-2")
+    assert ok is False and reason == "merge-lock busy"
+
+
+def test_tc11_stale_lease_is_reclaimed(lease_dir, monkeypatch):
+    repo = "orchestrator"
+    monkeypatch.setattr(merge_gate.settings, "merge_lock_timeout_s", 10)
+    # Write a lease that is older than the timeout (orphaned by a dead process).
+    path = merge_gate._lease_path(repo)
+    with open(path, "w", encoding="utf-8") as f:
+        json.dump(
+            {"branch": "feature/dead", "acquired_at": time.time() - 999, "pid": 1},
+            f,
+        )
+    ok, reason = merge_gate.acquire_merge_lease(repo, "feature/new", "ORCH-9")
+    assert ok is True
+    assert "reclaimed" in reason
+    # The new holder now owns it.
+    held = json.load(open(path, encoding="utf-8"))
+    assert held["branch"] == "feature/new"
+
+
+def test_tc11_release_missing_is_noop(lease_dir):
+    # Releasing a non-existent lease never raises.
+    merge_gate.release_merge_lease("orchestrator", "feature/none")
+    merge_gate.release_merge_lease("orchestrator")  # force form

tests/test_merge_gate_race.py

@@ -0,0 +1,150 @@
+"""ORCH-043 / TC-24: the parallel-merge race the gate exists to prevent.
+
+Scenario (two green branches in ONE repo, the self-hosting risk, ТЗ §1):
+  * main is at C1 because branch A already merged.
+  * branch B was validated against C0 (the main it branched from) and is GREEN
+    there — but B has NOT seen A's change. A blind merge of B could break main
+    (semantic conflict): B is "green" yet stale.
+
+The merge-gate makes this deterministic:
+  1. While A holds the merge-lease, B's gate sees "merge-lock busy" -> DEFER
+     (serialisation: no two catch-up+merge sequences interleave).
+  2. After A releases, B acquires the lease, rebases onto the CURRENT origin/main
+     (C1) and re-tests the COMBINED tree:
+       - re-test GREEN -> gate passes, lease HELD -> B is safe to merge; main stays green.
+       - re-test RED   -> gate fails, lease RELEASED -> B rolls back to development;
+                          main is NEVER touched.
+  origin/main's SHA is asserted unchanged throughout — the gate never pushes main.
+
+Real local git (bare origin + clone), real file lease; only the pytest re-test is
+stubbed (its real behaviour lives in test_merge_gate.py::retest_branch tests).
+"""
+import os
+import subprocess
+import tempfile
+
+import pytest
+
+_test_db = os.path.join(tempfile.gettempdir(), "test_orchestrator_merge_gate_race.db")
+os.environ["ORCH_DB_PATH"] = _test_db
+os.environ["ORCH_REPOS_DIR"] = tempfile.gettempdir()
+os.environ["ORCH_GITEA_TOKEN"] = "test-token"
+os.environ["ORCH_PLANE_API_TOKEN"] = "test-token"
+
+from src import git_worktree, merge_gate  # noqa: E402
+from src.qg import checks as qg  # noqa: E402
+from src.qg.checks import check_branch_mergeable  # noqa: E402
+
+
+def _git(cwd, *args):
+    return subprocess.run(["git", "-C", cwd, *args], capture_output=True, text=True)
+
+
+@pytest.fixture
+def race_repo(tmp_path, monkeypatch):
+    """Bare origin at C1 (A merged) + clone + feature/B branched from C0.
+
+    Returns (repo, origin_path). feature/B rebases cleanly onto origin/main.
+    The gate is forced REAL for this repo via merge_gate_repos.
+    """
+    repo = "orchestrator"
+    repos_dir = tmp_path / "repos"
+    wt_dir = tmp_path / "repos" / "_wt"
+    repos_dir.mkdir(parents=True)
+
+    monkeypatch.setattr(merge_gate.settings, "repos_dir", str(repos_dir))
+    monkeypatch.setattr(git_worktree.settings, "repos_dir", str(repos_dir))
+    monkeypatch.setattr(git_worktree.settings, "worktrees_dir", str(wt_dir))
+    monkeypatch.setattr(qg.settings, "merge_gate_enabled", True)
+    monkeypatch.setattr(qg.settings, "merge_gate_repos", repo)
+    monkeypatch.setattr(merge_gate.settings, "merge_lock_timeout_s", 300)
+
+    origin = tmp_path / "origin.git"
+    subprocess.run(["git", "init", "--bare", "-b", "main", str(origin)], capture_output=True)
+
+    seed = tmp_path / "seed"
+    seed.mkdir()
+    _git(str(seed), "init", "-b", "main")
+    _git(str(seed), "config", "user.email", "t@t")
+    _git(str(seed), "config", "user.name", "t")
+    (seed / "README.md").write_text("base\n")
+    _git(str(seed), "add", ".")
+    _git(str(seed), "commit", "-m", "C0")
+    _git(str(seed), "remote", "add", "origin", str(origin))
+    _git(str(seed), "push", "origin", "main")
+
+    # B branches off C0, adds an isolated file (clean rebase onto C1).
+    _git(str(seed), "checkout", "-b", "feature/B")
+    (seed / "b.txt").write_text("from B\n")
+    _git(str(seed), "add", ".")
+    _git(str(seed), "commit", "-m", "feat(B): add b.txt")
+    _git(str(seed), "push", "origin", "feature/B")
+
+    # A merged -> main advances to C1 (touches a DIFFERENT file: no textual conflict).
+    _git(str(seed), "checkout", "main")
+    (seed / "a.txt").write_text("from A\n")
+    _git(str(seed), "add", ".")
+    _git(str(seed), "commit", "-m", "C1 (A merged)")
+    _git(str(seed), "push", "origin", "main")
+
+    main_clone = repos_dir / repo
+    subprocess.run(["git", "clone", str(origin), str(main_clone)], capture_output=True)
+    _git(str(main_clone), "config", "user.email", "t@t")
+    _git(str(main_clone), "config", "user.name", "t")
+    return repo, origin
+
+
+def _origin_main_sha(origin):
+    return _git(str(origin), "rev-parse", "main").stdout.strip()
+
+
+def test_tc24_busy_lock_serialises_then_green_catch_up_is_safe(race_repo, monkeypatch):
+    """A holds the lease -> B defers; after release B catches up + green re-test ->
+    safe merge (lease held), and origin/main is never pushed by the gate."""
+    repo, origin = race_repo
+    main_before = _origin_main_sha(origin)
+
+    # A is mid-merge: it holds the lease.
+    ok, _ = merge_gate.acquire_merge_lease(repo, "feature/A", "ORCH-A")
+    assert ok is True
+
+    # B's gate must DEFER (serialisation), touching nothing.
+    passed, reason = check_branch_mergeable(repo, "ORCH-B", "feature/B")
+    assert passed is False
+    assert reason == "merge-lock busy"
+    assert _origin_main_sha(origin) == main_before  # main untouched
+
+    # A finishes and releases.
+    merge_gate.release_merge_lease(repo, "feature/A")
+
+    # B catches up: real rebase onto C1, GREEN re-test -> pass, lease HELD.
+    monkeypatch.setattr(merge_gate, "retest_branch", lambda r, b: (True, "re-test green"))
+    passed, reason = check_branch_mergeable(repo, "ORCH-B", "feature/B")
+    assert passed is True
+    assert reason == "rebased onto main, re-test green"
+    # The gate rebased+pushed ONLY the task branch; origin/main is unchanged.
+    assert _origin_main_sha(origin) == main_before
+    # feature/B now contains C1 (a.txt) on origin after the force-with-lease push.
+    assert "a.txt" in _git(str(origin), "ls-tree", "--name-only", "feature/B").stdout
+    # Lease is HELD by B until the actual merge.
+    held = merge_gate._read_lease(merge_gate._lease_path(repo))
+    assert held is not None and held.get("branch") == "feature/B"
+
+
+def test_tc24_red_catch_up_fails_and_releases_main_stays_green(race_repo, monkeypatch):
+    """B catches up but the COMBINED tree is red -> gate fails, lease released,
+    origin/main never touched (B will roll back to development upstream)."""
+    repo, origin = race_repo
+    main_before = _origin_main_sha(origin)
+
+    monkeypatch.setattr(
+        merge_gate, "retest_branch",
+        lambda r, b: (False, "re-test failed: ...1 failed, 9 passed"),
+    )
+    passed, reason = check_branch_mergeable(repo, "ORCH-B", "feature/B")
+    assert passed is False
+    assert reason.startswith("re-test failed after rebase:")
+    # main is still green / untouched.
+    assert _origin_main_sha(origin) == main_before
+    # The lease was released on failure (a later task can proceed).
+    assert merge_gate._read_lease(merge_gate._lease_path(repo)) is None

tests/test_qg_merge_gate.py

@@ -0,0 +1,211 @@
+"""ORCH-043 / TC-12..17: the merge-gate quality check ``check_branch_mergeable``.
+
+These exercise the COMPOSITION logic in src/qg/checks.check_branch_mergeable —
+the deterministic gate the engine runs on the deploy-staging -> deploy edge. The
+merge_gate primitives (rebase / re-test / lease) are mocked here; their real-git
+behaviour is covered in tests/test_merge_gate.py.
+
+Contract under test (ADR-001 §4):
+  * conditionality: merge_gate_enabled=False / repo-out-of-scope -> no-op pass,
+    NEVER touching the lease;
+  * up-to-date branch -> pass, lease HELD;
+  * behind + clean rebase + green re-test -> pass, lease HELD;
+  * rebase conflict -> fail, lease RELEASED;
+  * red / timeout re-test after rebase -> fail, lease RELEASED;
+  * never-raise: an exception inside the gate -> (False, ...) with lease released.
+"""
+
+import os
+
+os.environ.setdefault("ORCH_GITEA_TOKEN", "test-token")
+os.environ.setdefault("ORCH_PLANE_API_TOKEN", "test-token")
+
+import pytest  # noqa: E402
+
+from src import merge_gate  # noqa: E402
+from src.qg import checks as qg  # noqa: E402
+from src.qg.checks import check_branch_mergeable  # noqa: E402
+
+_REPO = "orchestrator"
+_BRANCH = "feature/ORCH-043-x"
+_WI = "ORCH-043"
+
+
+@pytest.fixture
+def lease_spy(monkeypatch):
+    """Replace the merge_gate lease primitives with in-memory spies.
+
+    Tracks acquire/release calls and lets each test program the acquire outcome
+    so we can assert the gate's lease lifecycle without touching the filesystem.
+    """
+    state = {
+        "acquired": False,
+        "released": False,
+        "acquire_result": (True, "lease acquired"),
+    }
+
+    def _acquire(repo, branch, work_item_id=None, task_id=None):
+        ok, reason = state["acquire_result"]
+        if ok:
+            state["acquired"] = True
+        return ok, reason
+
+    def _release(repo, branch=None):
+        state["released"] = True
+
+    monkeypatch.setattr(merge_gate, "acquire_merge_lease", _acquire)
+    monkeypatch.setattr(merge_gate, "release_merge_lease", _release)
+    # Default merge_gate scope: real for the self-hosting orchestrator repo.
+    monkeypatch.setattr(qg.settings, "merge_gate_enabled", True)
+    monkeypatch.setattr(qg.settings, "merge_gate_repos", "")
+    return state
+
+
+# ---------------------------------------------------------------------------
+# Conditionality (no-op variants) — must NOT touch the lease.
+# ---------------------------------------------------------------------------
+def test_tc16_disabled_is_noop(monkeypatch, lease_spy):
+    """TC-16 / AC-8: merge_gate_enabled=False -> pass, lease untouched."""
+    monkeypatch.setattr(qg.settings, "merge_gate_enabled", False)
+    ok, reason = check_branch_mergeable(_REPO, _WI, _BRANCH)
+    assert ok is True
+    assert reason == "merge-gate disabled"
+    assert lease_spy["acquired"] is False
+    assert lease_spy["released"] is False
+
+
+def test_tc17_repo_out_of_scope_is_noop(monkeypatch, lease_spy):
+    """TC-17 / AC-8: non-self-hosting repo (empty CSV) -> conditional no-op."""
+    ok, reason = check_branch_mergeable("enduro-trails", "ET-1", "feature/ET-1-x")
+    assert ok is True
+    assert reason == "merge-gate N/A for enduro-trails"
+    assert lease_spy["acquired"] is False
+    assert lease_spy["released"] is False
+
+
+def test_csv_scopes_gate_to_listed_repo(monkeypatch, lease_spy):
+    """merge_gate_repos CSV makes the gate real for a non-self-hosting repo."""
+    monkeypatch.setattr(qg.settings, "merge_gate_repos", "enduro-trails")
+    monkeypatch.setattr(merge_gate, "branch_is_behind_main", lambda r, b: False)
+    ok, reason = check_branch_mergeable("enduro-trails", "ET-1", "feature/ET-1-x")
+    assert ok is True
+    assert reason == "branch up-to-date with main"
+    assert lease_spy["acquired"] is True  # gate actually ran
+
+
+# ---------------------------------------------------------------------------
+# Lock busy -> DEFER signal (no rollback at this layer).
+# ---------------------------------------------------------------------------
+def test_lock_busy_returns_defer_signal(monkeypatch, lease_spy):
+    """Lease busy -> (False, 'merge-lock busy'); nothing acquired or released."""
+    lease_spy["acquire_result"] = (False, "merge-lock busy")
+    ok, reason = check_branch_mergeable(_REPO, _WI, _BRANCH)
+    assert ok is False
+    assert reason == "merge-lock busy"
+    assert lease_spy["acquired"] is False
+    assert lease_spy["released"] is False  # we never held it
+
+
+# ---------------------------------------------------------------------------
+# TC-12: branch already up-to-date -> pass, lease HELD.
+# ---------------------------------------------------------------------------
+def test_tc12_up_to_date_passes_lease_held(monkeypatch, lease_spy):
+    monkeypatch.setattr(merge_gate, "branch_is_behind_main", lambda r, b: False)
+    # If these were called the test would wrongly proceed — guard with raisers.
+    monkeypatch.setattr(
+        merge_gate, "auto_rebase_onto_main",
+        lambda r, b: pytest.fail("must not rebase an up-to-date branch"),
+    )
+    ok, reason = check_branch_mergeable(_REPO, _WI, _BRANCH)
+    assert ok is True
+    assert reason == "branch up-to-date with main"
+    assert lease_spy["acquired"] is True
+    assert lease_spy["released"] is False  # lease HELD until the merge
+
+
+# ---------------------------------------------------------------------------
+# TC-13: behind + clean rebase + green re-test -> pass, lease HELD.
+# ---------------------------------------------------------------------------
+def test_tc13_behind_clean_rebase_green_passes_lease_held(monkeypatch, lease_spy):
+    monkeypatch.setattr(merge_gate, "branch_is_behind_main", lambda r, b: True)
+    monkeypatch.setattr(
+        merge_gate, "auto_rebase_onto_main",
+        lambda r, b: (True, "rebased onto origin/main"),
+    )
+    monkeypatch.setattr(merge_gate, "retest_branch", lambda r, b: (True, "re-test green"))
+    ok, reason = check_branch_mergeable(_REPO, _WI, _BRANCH)
+    assert ok is True
+    assert reason == "rebased onto main, re-test green"
+    assert lease_spy["acquired"] is True
+    assert lease_spy["released"] is False  # lease HELD
+
+
+# ---------------------------------------------------------------------------
+# TC-14: rebase conflict -> fail, lease RELEASED.
+# ---------------------------------------------------------------------------
+def test_tc14_rebase_conflict_fails_lease_released(monkeypatch, lease_spy):
+    monkeypatch.setattr(merge_gate, "branch_is_behind_main", lambda r, b: True)
+    monkeypatch.setattr(
+        merge_gate, "auto_rebase_onto_main",
+        lambda r, b: (False, "rebase conflict: src/db.py"),
+    )
+    monkeypatch.setattr(
+        merge_gate, "retest_branch",
+        lambda r, b: pytest.fail("must not re-test after a failed rebase"),
+    )
+    ok, reason = check_branch_mergeable(_REPO, _WI, _BRANCH)
+    assert ok is False
+    assert reason == "rebase conflict: src/db.py"
+    assert lease_spy["released"] is True
+
+
+# ---------------------------------------------------------------------------
+# TC-15: red / timeout re-test after rebase -> fail, lease RELEASED.
+# ---------------------------------------------------------------------------
+def test_tc15_red_retest_fails_lease_released(monkeypatch, lease_spy):
+    monkeypatch.setattr(merge_gate, "branch_is_behind_main", lambda r, b: True)
+    monkeypatch.setattr(
+        merge_gate, "auto_rebase_onto_main",
+        lambda r, b: (True, "rebased onto origin/main"),
+    )
+    monkeypatch.setattr(
+        merge_gate, "retest_branch",
+        lambda r, b: (False, "re-test failed: ...1 failed, 5 passed"),
+    )
+    ok, reason = check_branch_mergeable(_REPO, _WI, _BRANCH)
+    assert ok is False
+    assert reason.startswith("re-test failed after rebase:")
+    assert "1 failed, 5 passed" in reason
+    assert lease_spy["released"] is True
+
+
+def test_tc15_retest_timeout_passes_reason_through(monkeypatch, lease_spy):
+    """AC-6: a re-test timeout keeps its distinct reason and releases the lease."""
+    monkeypatch.setattr(merge_gate, "branch_is_behind_main", lambda r, b: True)
+    monkeypatch.setattr(
+        merge_gate, "auto_rebase_onto_main",
+        lambda r, b: (True, "rebased onto origin/main"),
+    )
+    monkeypatch.setattr(
+        merge_gate, "retest_branch",
+        lambda r, b: (False, "re-test timeout after 600s"),
+    )
+    ok, reason = check_branch_mergeable(_REPO, _WI, _BRANCH)
+    assert ok is False
+    assert reason == "re-test timeout after 600s"
+    assert lease_spy["released"] is True
+
+
+# ---------------------------------------------------------------------------
+# Never-raise: an exception inside the gate -> (False, ...) + lease released.
+# ---------------------------------------------------------------------------
+def test_never_raise_releases_lease_on_internal_error(monkeypatch, lease_spy):
+    """AC-9: a blowing-up primitive is caught; the gate returns and releases."""
+    def _boom(r, b):
+        raise RuntimeError("git exploded")
+
+    monkeypatch.setattr(merge_gate, "branch_is_behind_main", _boom)
+    ok, reason = check_branch_mergeable(_REPO, _WI, _BRANCH)
+    assert ok is False
+    assert "merge-gate error" in reason
+    assert lease_spy["released"] is True  # held then released on the error path

tests/test_qg_registry_snapshot.py

@@ -28,6 +28,7 @@ _EXPECTED_QGS = {
     "check_tests_local",
     "check_deploy_status",
     "check_staging_status",
+    "check_branch_mergeable",  # ORCH-043 merge-gate (deploy-staging -> deploy edge)
 }
 
 

tests/test_stage_engine.py

@@ -805,6 +805,188 @@ class TestStagingGate:
 # ---------------------------------------------------------------------------
 # launcher + plane both delegate to the engine
 # ---------------------------------------------------------------------------
+class TestMergeGate:
+    """ORCH-043 / TC-20..23: the merge-gate sub-gate on the deploy-staging -> deploy
+    edge. The QG ``check_branch_mergeable`` is monkeypatched on stage_engine.QG_CHECKS
+    so we drive the engine's reaction (advance / defer / rollback) deterministically;
+    the gate's own composition is covered in test_qg_merge_gate.py.
+    """
+
+    def _jobs_full(self):
+        conn = get_db()
+        rows = conn.execute(
+            "SELECT agent, task_content, available_at FROM jobs ORDER BY id"
+        ).fetchall()
+        conn.close()
+        return [dict(r) for r in rows]
+
+    def test_tc20_pass_advances_to_deploy(self, monkeypatch):
+        """TC-20 / AC-1: gate PASS (rebased + green) -> advance to deploy, deployer
+        enqueued, NO rollback. staging gate must pass first (same edge)."""
+        monkeypatch.setattr(
+            stage_engine, "QG_CHECKS",
+            {**stage_engine.QG_CHECKS,
+             "check_staging_status": _pass,
+             "check_branch_mergeable": _pass},
+        )
+        task_id = _make_task("deploy-staging", repo="orchestrator", wi="ORCH-043",
+                             branch="feature/ORCH-043-x")
+        res = advance_stage(
+            task_id, "deploy-staging", "orchestrator", "ORCH-043",
+            "feature/ORCH-043-x", finished_agent="deployer",
+        )
+        assert res.advanced is True
+        assert res.to_stage == "deploy"
+        assert _stage(task_id) == "deploy"
+        assert res.rolled_back_to is None
+        jobs = _jobs()
+        assert len(jobs) == 1
+        assert jobs[0]["agent"] == "deployer"
+
+    def test_tc21_busy_lock_defers_without_rollback(self, monkeypatch):
+        """TC-21 / AC-5: 'merge-lock busy' -> DEFER: task stays on deploy-staging,
+        deployer re-queued with a delay (available_at set), no rollback, no alert."""
+        monkeypatch.setattr(
+            stage_engine, "QG_CHECKS",
+            {**stage_engine.QG_CHECKS,
+             "check_staging_status": _pass,
+             "check_branch_mergeable": _fail("merge-lock busy")},
+        )
+        monkeypatch.setattr(stage_engine.settings, "merge_defer_delay_s", 30)
+        monkeypatch.setattr(stage_engine.settings, "merge_defer_max_attempts", 5)
+        task_id = _make_task("deploy-staging", repo="orchestrator", wi="ORCH-043",
+                             branch="feature/ORCH-043-x")
+        res = advance_stage(
+            task_id, "deploy-staging", "orchestrator", "ORCH-043",
+            "feature/ORCH-043-x", finished_agent="deployer",
+        )
+        assert res.advanced is False
+        assert res.rolled_back_to is None
+        assert res.note == "merge-gate-deferred"
+        assert _stage(task_id) == "deploy-staging"   # stays put
+        jobs = self._jobs_full()
+        assert len(jobs) == 1
+        assert jobs[0]["agent"] == "deployer"
+        assert "merge-gate defer" in jobs[0]["task_content"]
+        assert jobs[0]["available_at"] is not None   # delayed re-pickup
+        assert stage_engine.set_issue_blocked.called is False
+
+    def test_tc21_defer_exhausted_blocks_and_alerts(self, monkeypatch):
+        """AC-5: after merge_defer_max_attempts defers -> block + Telegram, no new job."""
+        monkeypatch.setattr(
+            stage_engine, "QG_CHECKS",
+            {**stage_engine.QG_CHECKS,
+             "check_staging_status": _pass,
+             "check_branch_mergeable": _fail("merge-lock busy")},
+        )
+        monkeypatch.setattr(stage_engine.settings, "merge_defer_max_attempts", 3)
+        task_id = _make_task("deploy-staging", repo="orchestrator", wi="ORCH-043",
+                             branch="feature/ORCH-043-x")
+        # Pre-seed 3 prior defer jobs (the restart-safe counter reads task_content).
+        conn = get_db()
+        for _ in range(3):
+            conn.execute(
+                "INSERT INTO jobs (agent, repo, task_id, task_content) "
+                "VALUES ('deployer','orchestrator',?, 'Note: merge-gate defer')",
+                (task_id,),
+            )
+        conn.commit()
+        conn.close()
+        res = advance_stage(
+            task_id, "deploy-staging", "orchestrator", "ORCH-043",
+            "feature/ORCH-043-x", finished_agent="deployer",
+        )
+        assert res.advanced is False
+        assert res.note == "merge-gate-defer-exhausted"
+        assert res.alerted is True
+        assert stage_engine.set_issue_blocked.called
+        assert stage_engine.send_telegram.called
+        # No NEW defer job past the cap (still the 3 we seeded).
+        assert len(self._jobs_full()) == 3
+
+    def test_tc22_conflict_rolls_back_to_development(self, monkeypatch):
+        """TC-22 / AC-3: rebase conflict -> rollback to development + developer retry."""
+        monkeypatch.setattr(
+            stage_engine, "QG_CHECKS",
+            {**stage_engine.QG_CHECKS,
+             "check_staging_status": _pass,
+             "check_branch_mergeable": _fail("rebase conflict: src/db.py")},
+        )
+        task_id = _make_task("deploy-staging", repo="orchestrator", wi="ORCH-043",
+                             branch="feature/ORCH-043-x")
+        res = advance_stage(
+            task_id, "deploy-staging", "orchestrator", "ORCH-043",
+            "feature/ORCH-043-x", finished_agent="deployer",
+        )
+        assert res.advanced is False
+        assert res.rolled_back_to == "development"
+        assert _stage(task_id) == "development"
+        assert res.qg_name == "check_branch_mergeable"
+        jobs = _jobs()
+        assert len(jobs) == 1
+        assert jobs[0]["agent"] == "developer"
+        assert stage_engine.set_issue_in_progress.called
+
+    def test_tc22_red_retest_rolls_back_to_development(self, monkeypatch):
+        """AC-2/AC-3: red re-test after rebase -> rollback to development."""
+        monkeypatch.setattr(
+            stage_engine, "QG_CHECKS",
+            {**stage_engine.QG_CHECKS,
+             "check_staging_status": _pass,
+             "check_branch_mergeable": _fail("re-test failed after rebase: 1 failed")},
+        )
+        task_id = _make_task("deploy-staging", repo="orchestrator", wi="ORCH-043",
+                             branch="feature/ORCH-043-x")
+        res = advance_stage(
+            task_id, "deploy-staging", "orchestrator", "ORCH-043",
+            "feature/ORCH-043-x", finished_agent="deployer",
+        )
+        assert res.rolled_back_to == "development"
+        assert _stage(task_id) == "development"
+        jobs = _jobs()
+        assert len(jobs) == 1
+        assert jobs[0]["agent"] == "developer"
+        # The rollback task_desc carries the gate reason for the developer.
+        assert "re-test failed after rebase: 1 failed" in _job_contents()[0]
+
+    def test_tc23_rollback_respects_max_developer_retries(self, monkeypatch):
+        """TC-23 / AC-11: merge-gate rollback is capped by MAX_DEVELOPER_RETRIES —
+        no infinite bounce. 4th attempt -> block + alert, no new developer job."""
+        monkeypatch.setattr(
+            stage_engine, "QG_CHECKS",
+            {**stage_engine.QG_CHECKS,
+             "check_staging_status": _pass,
+             "check_branch_mergeable": _fail("rebase conflict: src/db.py")},
+        )
+        task_id = _make_task("deploy-staging", repo="orchestrator", wi="ORCH-043",
+                             branch="feature/ORCH-043-x")
+        _add_developer_runs(task_id, 3)  # already at the cap
+        res = advance_stage(
+            task_id, "deploy-staging", "orchestrator", "ORCH-043",
+            "feature/ORCH-043-x", finished_agent="deployer",
+        )
+        assert res.rolled_back_to == "development"
+        assert stage_engine.set_issue_blocked.called
+        assert stage_engine.send_telegram.called
+        assert _jobs() == []   # no developer job past the cap
+
+    def test_non_self_hosting_repo_skips_merge_gate(self, monkeypatch):
+        """Regression: for a non-self-hosting repo the REAL gate is a no-op, so
+        deploy-staging -> deploy advances exactly as before ORCH-043."""
+        monkeypatch.setattr(
+            stage_engine, "QG_CHECKS",
+            {**stage_engine.QG_CHECKS, "check_staging_status": _pass},
+        )  # check_branch_mergeable left REAL -> N/A for enduro-trails
+        task_id = _make_task("deploy-staging", repo="enduro-trails", wi="ET-035",
+                             branch="feature/ET-035-x")
+        res = advance_stage(
+            task_id, "deploy-staging", "enduro-trails", "ET-035",
+            "feature/ET-035-x", finished_agent="deployer",
+        )
+        assert res.advanced is True
+        assert _stage(task_id) == "deploy"
+
+
 class TestDelegation:
     def test_launcher_calls_engine(self):
         from src.agents.launcher import AgentLauncher