fix(reaper): do not re-run deploy-staging finalization while finalizer is alive

On the deploy-staging -> deploy edge the live monitor stamps
agent_runs.finished_at FIRST, then runs the heavy edge sub-gates
(security/merge-gate re-test/coverage/image-freshness) in-thread for MINUTES
and only THEN _finalize_job. Reaper Tier-2 measures finished_age_s from
finished_at, so past reaper_finalize_grace_s it treated the live, long
finalizer as dead and independently re-ran the advance -> a second re-test
went red -> false rollback deploy-staging -> development while the original
finalizer concurrently merged the PR (incident ORCH-111, job 1914).

Add a process-local finalizer-ownership registry (src/finalizer_liveness.py,
never-raise): the monitor mark()s ownership right after the exit_code stamp and
clear()s it in a try/finally around the (verbatim-extracted) finalization tail,
so an exception in the monitor thread still releases ownership and a genuinely
dead finalizer is reaped. The reaper Tier-2 consults the marker only when the
kill-switch is on AND the task stage == deploy-staging AND ownership is active
-> DEFER (no second advance) and fall through to the Tier-3 backstop, which
ignores the marker (a stuck/dead finalizer is still reaped in bounded time).
In-memory is authoritative (monitor + reaper are daemon threads of one uvicorn
process); restart is covered by the startup requeue_running_jobs.

Additive, global kill-switch reaper_finalizer_liveness_enabled (default True;
false -> reaper byte-for-byte prior). STAGE_TRANSITIONS / QG_CHECKS / every
check_* / machine-verdict keys / DB schema unchanged; grace/ceiling and the
ORCH-065/109/110 budget invariant untouched; never restarts prod, never pushes
main. Observability: finalizer_defers_total + finalizer_owned in GET /queue.
Tests: tests/test_orch113_reaper_finalizer_liveness.py (TC-01..TC-08, incl. the
mandatory ORCH-111 regression: red before the fix, green after).

Refs: ORCH-113

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

CHANGELOG.md

@@ -3,6 +3,11 @@
 Формат: [Keep a Changelog](https://keepachangelog.com/). Записи — на смысловой PR/задачу.
 
 ## [Unreleased]
+- **Job-reaper не реапит живой долго финализирующий монитор `deploy-staging`** (ORCH-113, `fix`, bug→escalate full-cycle): устранено расхождение состояния из инцидента ORCH-111 (deployer job 1914 / run_id 683). На ребре `deploy-staging → deploy` живой монитор (`launcher._monitor_agent`) штампит `agent_runs.finished_at`/`exit_code` **первым**, затем синхронно в своём потоке прогоняет тяжёлые edge-под-гейты (`security → merge-gate re-test → coverage → image-freshness`) — **минуты** — и лишь потом `_finalize_job`. Reaper Tier-2 меряет `finished_age_s` от `finished_at` (= начала финализации), поэтому по истечении `reaper_finalize_grace_s=300` трактовал живого долго финализирующего монитора как мёртвого и **независимо** повторял тот же тяжёлый advance: повторный re-test стал красным → ложный откат `deploy-staging → development` (+ ложный developer-retry) **параллельно** с тем, что исходный finalizer довёл deploy до SUCCESS и смержил PR — состояние раздвоилось. Аддитивно, под глобальным kill-switch, never-raise; `STAGE_TRANSITIONS`/`QG_CHECKS`/каждый `check_*`/machine-verdict ключи/схема БД — **байт-в-байт не тронуты**; `reaper_finalize_grace_s`/`reaper_max_running_s` и сквозной бюджет ORCH-065/109/110 (`5400 > Σ(gate-work)+grace`) сохранены; фикс не рестартит прод и не пушит `main`. ADR: `docs/work-items/ORCH-113/06-adr/ADR-001-reaper-finalizer-liveness-ownership.md`, сквозной `docs/architecture/adr/adr-0043-reaper-finalizer-liveness-ownership.md`.
+  - **Leaf `src/finalizer_liveness.py` (новый, процесс-локальный реестр владения):** чистый never-raise модуль (паттерн `serial_gate`/`coverage_gate`, без сети/БД) — `mark(job_id, run_id, stage)` / `clear(job_id)` / `is_active(job_id)` / `snapshot()`; состояние `{job_id: {...}}` под `threading.Lock`. Авторитетно in-memory, т.к. монитор и reaper — daemon-**потоки одного** uvicorn-процесса (CMD без `--workers`) с общей SQLite-БД. Собственного TTL нет — ограничение по времени даёт Tier-3 backstop. `is_active` при ошибке → `False` (консервативно: не блокировать добивание).
+  - **Эмиссия владения (`launcher._monitor_agent`):** `mark()` вызывается **сразу после** штампа `exit_code` (самый ранний момент Tier-2), хвост финализации вынесен в `_run_monitor_finalization` и обёрнут в `try/finally` с `clear()` в `finally` → исключение в потоке монитора гарантированно снимает владение, и реально мёртвый finalizer добивается. Маркер пишется **безусловно** (kill-switch гейтит только консультацию reaper, поэтому выключенный путь — байт-в-байт прежний). Хвост перенесён **дословно** (проверяется `git diff -w`: +49/−0, нулевое изменение логики).
+  - **Консультация reaper (`job_reaper._reap_job` Tier-2):** при `reaper_finalizer_liveness_enabled` **И** стадии задачи `== "deploy-staging"` **И** активном владении → **defer** (счётчик + лог, не повторять advance), провал к Tier-3. **Tier-3 (`age >= reaper_max_running_s`) маркер игнорирует** — застрявший/мёртвый finalizer добивается в ограниченное время. Скоуп — только глобальный kill-switch `reaper_finalizer_liveness_enabled` (env `ORCH_REAPER_FINALIZER_LIVENESS_ENABLED`, дефолт `True`; `False` → reaper байт-в-байт прежний), **без** per-repo разреза (баг общий для всех репо со стадией `deploy-staging`).
+  - **Наблюдаемость:** аддитивные ключи `finalizer_liveness_enabled`/`finalizer_defers_total`/`finalizer_owned` в блоке `reaper` ответа `GET /queue` (существующие ключи не тронуты). Покрытие — `tests/test_orch113_reaper_finalizer_liveness.py` (TC-01…TC-08, включая обязательный регресс ORCH-111: КРАСНЫЙ до фикса, ЗЕЛЁНЫЙ после).
 - **Merge-gate re-test: толерантность к инфра-таймауту + tree-kill спавненных pytest + контракт необходимости re-test** (ORCH-110, `fix`, bug→escalate full-cycle): устранён ложный откат `deploy-staging → development`, возникавший когда локальный re-test merge-gate падал по **таймауту** (инфра/ресурс) при зелёных CI + tester + staging (инцидент ORCH-109/PR #129: сюит 516.7s упёрся в бюджет 600s под CPU-голоданием от осиротевших pytest-процессов → `(False, "re-test timeout after 600s")` → `_handle_merge_gate_rollback` → каждый из 3 developer-retry падал так же → «Merge-gate still failing after 3 developer retries» → ручное вмешательство). Аддитивно, под 5 независимыми kill-switch, never-raise, скоуп self-hosting; `STAGE_TRANSITIONS`/реестр `QG_CHECKS`/семантика `check_*`/machine-verdict-ключи/схема БД — **байт-в-байт не тронуты**; INV-4 (никогда push/force-push `main`) и запрет рестарта прод-контейнера — соблюдены. ADR: `docs/work-items/ORCH-110/06-adr/ADR-001-merge-gate-retest-infra-tolerance-and-tree-kill.md`, сквозной `docs/architecture/adr/adr-0042-merge-gate-retest-infra-tolerance-and-tree-kill.md`.
   - **D1 — process-group tree-kill (`src/proc_group.py`, новый stdlib-only leaf):** `merge_gate.retest_branch` и `coverage_gate.measure_coverage` теперь спавнят pytest в **отдельной группе процессов** (`start_new_session`) и при таймауте убивают **всё дерево** (`os.killpg`, каскад SIGTERM→grace→SIGKILL по образцу `launcher.stop_process`), а не только прямого потомка — осиротевшие внуки-pytest больше не переживают бюджет и не грузят CPU. Контракты возврата сохранены (меняется лишь побочный эффект — нет утечки). Грейс реюзит `agent_kill_grace_seconds`. Fallback never-break: `subprocess_tree_kill_enabled=False` или не-POSIX → прежний `subprocess.run(timeout=)`.
   - **D2/D3 — классификация + маршрутизация инфра-таймаута:** чистый предикат `merge_gate.classify_retest_failure(reason)` различает `timeout`/`red`/`lock-busy`/`other` (scope-guard: `auto_rebase_onto_main`'s «rebase timeout» — НЕ инфра-таймаут re-test, остаётся на rollback-пути). Инфра-таймаут → новый `_handle_merge_gate_infra_retry` (ограниченный повтор по образцу `_handle_merge_gate_defer`: задача **остаётся на deploy-staging**, staging-deployer перезапускается с задержкой, **БЕЗ** отката на `development` и **БЕЗ** расхода developer-retry). Анти-над-толерантность (BR-6): детерминированно **красный** re-test / конфликт по-прежнему → `_handle_merge_gate_rollback`. Anti-loop: исчерпание бюджета → один **инфра-alert** (явно инфраструктурная формулировка «НЕ дефект кода» с кликабельным номером), задача НЕ уходит в `development`.

src/agents/launcher.py

@@ -868,6 +868,55 @@ class AgentLauncher:
         _task_id = _row[0] if _row else None
         conn.close()
 
+        # ORCH-113 (adr-0043 / D2): register finalizer ownership at the EARLIEST
+        # moment the reaper can enter Tier-2 — exit_code is now stamped, so the agent
+        # pid is dead and Tier-1 no longer protects this row. On the
+        # deploy-staging -> deploy edge the rest of the finalization (git push, the
+        # heavy edge sub-gates via _try_advance_stage, _finalize_job) runs IN THIS
+        # THREAD for MINUTES; without ownership the reaper would treat the live
+        # finalizer as dead and re-run the advance (false rollback, incident
+        # ORCH-111). The marker is written unconditionally (kill-switch only gates the
+        # reaper's CONSULTATION, so the disabled path is byte-for-byte prior). Only
+        # queue-launched jobs (job_id is not None) are in get_running_jobs / reapable.
+        if job_id is not None:
+            try:
+                from .. import finalizer_liveness
+                _t_mark = get_task_by_repo_branch(repo, branch)
+                finalizer_liveness.mark(
+                    job_id, run_id, _t_mark["stage"] if _t_mark else None
+                )
+            except Exception:  # noqa: BLE001 - never-raise: marker is best-effort
+                pass
+
+        # The finalization tail runs under try/finally so ownership is ALWAYS
+        # released — including on ANY exception in this thread, which lets the reaper
+        # finish a genuinely dead finalizer (FR-4).
+        try:
+            self._run_monitor_finalization(
+                run_id, agent, repo, branch, exit_code, output_path,
+                job_id, _task_id, _duration_s,
+            )
+        finally:
+            if job_id is not None:
+                try:
+                    from .. import finalizer_liveness
+                    finalizer_liveness.clear(job_id)
+                except Exception:  # noqa: BLE001 - never-raise
+                    pass
+
+    def _run_monitor_finalization(
+        self, run_id, agent, repo, branch, exit_code, output_path,
+        job_id, _task_id, _duration_s,
+    ):
+        """ORCH-113 (adr-0043): finalization tail of ``_monitor_agent``.
+
+        Extracted VERBATIM from ``_monitor_agent`` (no logic change — verify with
+        ``git diff -w``) so the caller can wrap it in a single ``try/finally`` that
+        releases the finalizer-ownership marker (``finalizer_liveness.clear``) on any
+        outcome. Runs in the monitor thread: notify, usage accounting, git
+        commit/push (+PR), failure handling, usage comments, the gate-driven
+        ``_try_advance_stage`` and finally ``_finalize_job`` for queue-launched jobs.
+        """
         notify_agent_finished(run_id, agent, exit_code, task_id=_task_id, duration_s=_duration_s)
 
         # Feature 4: parse token usage / cost from the (json) run log and record

src/config.py

@@ -544,12 +544,32 @@ class Settings(BaseSettings):
     #   lease_reclaim_enabled -> kill-switch for the proactive stale/dead lease reclaim
     #                           (false -> only the legacy lazy TTL reclaim in acquire).
     # (reuse) merge_lock_timeout_s -> lease TTL; merge_gate_repos -> reclaim scope.
+    #   reaper_finalizer_liveness_enabled -> ORCH-113 (adr-0043): consult the
+    #                           process-local finalizer-ownership registry
+    #                           (src/finalizer_liveness.py) in the Tier-2 branch. On the
+    #                           deploy-staging -> deploy edge the monitor runs the heavy
+    #                           edge sub-gates (security/merge-gate re-test/coverage/
+    #                           image-freshness) in-thread AFTER the finished_at stamp and
+    #                           BEFORE _finalize_job — MINUTES — which finished_age_s grace
+    #                           does NOT cover, so a live long finalizer was wrongly reaped
+    #                           and re-ran the advance (false rollback, incident ORCH-111).
+    #                           When true AND the task stage == 'deploy-staging' AND a live
+    #                           monitor owns the finalization, the reaper DEFERS (no second
+    #                           advance) and falls through to the Tier-3 backstop (which
+    #                           ignores the marker -> a stuck/dead finalizer is still reaped
+    #                           in bounded time). false -> reaper byte-for-byte as before
+    #                           (the marker is still written by the monitor but never
+    #                           consulted -> inert). Global only, NO per-repo split (the bug
+    #                           is common to every repo with a deploy-staging stage). Does
+    #                           NOT touch grace/ceiling -> the cross-cutting budget
+    #                           reaper_max_running_s > Σ(gate-work) + grace stays intact.
     reaper_enabled: bool = True
     reaper_interval_s: int = 60
     reaper_dead_ticks: int = 2
     reaper_max_running_s: int = 5400
     reaper_finalize_grace_s: int = 300
     lease_reclaim_enabled: bool = True
+    reaper_finalizer_liveness_enabled: bool = True
 
     # ORCH-063: disk-watchdog — background heartbeat that measures host-FS fill via
     # the mounted bind-paths and Telegram-alerts the operator at >= threshold. On

src/finalizer_liveness.py

@@ -0,0 +1,120 @@
+"""ORCH-113 (adr-0043): process-local finalizer-ownership registry.
+
+Leaf module — pure, process-local, never-raise (pattern of ``serial_gate`` /
+``coverage_gate``: imports nothing from ``stage_engine`` / ``launcher`` / the DB,
+talks to no network). It records "a LIVE monitor thread is currently finalizing
+job X" so the job-reaper can tell a long-running-but-alive finalizer apart from a
+genuinely dead one.
+
+Why in-memory is authoritative (ADR-001 / adr-0043): the monitor
+(``launcher._monitor_agent``) and the reaper (``job_reaper``) are daemon THREADS
+of the SAME single uvicorn process (CMD has no ``--workers``), sharing one SQLite
+DB. So liveness of the finalizing thread can be observed in-process. A whole-process
+death is covered by the startup ``requeue_running_jobs()`` (``running -> queued``),
+which ``main.lifespan`` runs BEFORE the reaper starts — so a restart leaves this
+registry empty and the requeued jobs are re-driven cleanly (restart-safe, no durable
+state needed).
+
+The bug this closes (incident ORCH-111, deployer job 1914): on the
+``deploy-staging -> deploy`` edge the monitor stamps ``agent_runs.finished_at``
+FIRST, then runs the heavy edge sub-gates (security -> merge-gate re-test ->
+coverage -> image-freshness) synchronously in its own thread — MINUTES — and only
+THEN ``_finalize_job``. Reaper Tier-2 measures ``finished_age_s`` from
+``finished_at`` (= the START of finalization), so once it exceeds
+``reaper_finalize_grace_s`` (300s) it treated the live, long-finalizing monitor as
+dead and independently re-ran the same heavy advance -> a second re-test went red ->
+false rollback ``deploy-staging -> development`` while the original finalizer
+concurrently merged the PR. State diverged.
+
+No own TTL: time-bounding is the reaper's Tier-3 backstop (``reaper_max_running_s``),
+which deliberately IGNORES this marker so a truly stuck finalizer is still reaped in
+bounded time. Every public function is isolated (``try/except`` -> safe default);
+``is_active`` defaults to ``False`` on error (conservative: never block the reaping
+of a possibly-dead finalizer).
+
+See docs/work-items/ORCH-113/06-adr/ADR-001-reaper-finalizer-liveness-ownership.md
+and the cross-cutting docs/architecture/adr/adr-0043-reaper-finalizer-liveness-ownership.md.
+"""
+from __future__ import annotations
+
+import logging
+import threading
+import time
+
+logger = logging.getLogger("orchestrator.finalizer_liveness")
+
+# Process-local ownership registry: {job_id: {"run_id", "stage", "started_ts"}}.
+# Guarded by a Lock because the monitor thread writes (mark/clear) while the reaper
+# thread reads (is_active/snapshot). All state resets on process restart, which is
+# safe (the startup requeue_running_jobs covers the restart path).
+_LOCK = threading.Lock()
+_OWNED: dict[int, dict] = {}
+
+
+def mark(job_id: int | None, run_id: int | None, stage: str | None) -> None:
+    """Register that a live monitor thread is finalizing ``job_id``.
+
+    Called by ``launcher._monitor_agent`` right after the ``exit_code`` stamp (the
+    earliest moment the reaper can enter Tier-2). ``stage`` is best-effort context
+    for the snapshot only — the reaper decides the actual stage from ``tasks`` via
+    its own ``_task_meta`` lookup. No-op when ``job_id is None`` (legacy direct
+    ``launch()`` jobs are not in ``get_running_jobs`` and are unreapable). Never
+    raises.
+    """
+    if job_id is None:
+        return
+    try:
+        with _LOCK:
+            _OWNED[job_id] = {
+                "run_id": run_id,
+                "stage": stage,
+                "started_ts": time.time(),
+            }
+    except Exception as e:  # noqa: BLE001 - never-raise contract
+        logger.warning("finalizer_liveness.mark failed for job %s: %s", job_id, e)
+
+
+def clear(job_id: int | None) -> None:
+    """Release ownership of ``job_id`` (idempotent).
+
+    Called from the ``finally`` of the monitor's finalization tail, so ANY exception
+    in the monitor thread still releases ownership -> a genuinely dead finalizer is
+    reaped (FR-4). Never raises.
+    """
+    if job_id is None:
+        return
+    try:
+        with _LOCK:
+            _OWNED.pop(job_id, None)
+    except Exception as e:  # noqa: BLE001 - never-raise contract
+        logger.warning("finalizer_liveness.clear failed for job %s: %s", job_id, e)
+
+
+def is_active(job_id: int | None) -> bool:
+    """True iff a live monitor currently owns the finalization of ``job_id``.
+
+    Consulted by the reaper Tier-2 branch. Defaults to ``False`` on any error or
+    when ``job_id is None`` (conservative: never block the reaping of a possibly
+    dead finalizer). Never raises.
+    """
+    if job_id is None:
+        return False
+    try:
+        with _LOCK:
+            return job_id in _OWNED
+    except Exception as e:  # noqa: BLE001 - never-raise contract
+        logger.warning("finalizer_liveness.is_active failed for job %s: %s", job_id, e)
+        return False
+
+
+def snapshot() -> dict:
+    """Read-only view of current ownership for ``GET /queue`` observability.
+
+    Returns ``{"active": <count>, "jobs": [job_id, ...]}``. Never raises.
+    """
+    try:
+        with _LOCK:
+            return {"active": len(_OWNED), "jobs": sorted(_OWNED.keys())}
+    except Exception as e:  # noqa: BLE001 - never-raise contract
+        logger.warning("finalizer_liveness.snapshot failed: %s", e)
+        return {"active": 0, "jobs": []}

src/job_reaper.py

@@ -38,7 +38,15 @@ Liveness (defense in depth, ADR-001 Р-1):
     reaper therefore treats it as a dead monitor (KNOWN outcome) only after a
     finalization grace: ``exit_code`` recorded for >= ``reaper_finalize_grace_s``
     (a live finalizing monitor is NEVER reaped, FR-1.3/AC-3). Within the grace the
-    row is left untouched.
+    row is left untouched. **ORCH-113 (adr-0043):** on the ``deploy-staging ->
+    deploy`` edge the in-thread finalization runs the heavy edge sub-gates
+    (security/merge-gate re-test/coverage/image-freshness) for MINUTES AFTER the
+    ``finished_at`` stamp, so even past the grace the monitor may be alive. Tier-2
+    now consults a process-local ownership marker (``finalizer_liveness``): a job
+    on ``deploy-staging`` still owned by a live finalizer is DEFERRED (not reaped via
+    Tier-2 — re-running the advance caused the false rollback in incident ORCH-111)
+    and falls through to the Tier-3 backstop, which IGNORES the marker. Kill-switch
+    ``reaper_finalizer_liveness_enabled``.
   * **Tier-3 (backstop): age ceiling.** A job ``running`` longer than
     ``reaper_max_running_s`` (deliberately > max ``agent_timeout`` + grace) is
     reaped even when liveness cannot be determined (pid reused / unknown).
@@ -142,6 +150,10 @@ class JobReaper:
         self.reaped_total: int = 0
         self.last_reaped: dict | None = None
         self.lease_reclaimed_total: int = 0
+        # ORCH-113 (adr-0043 / D5): count of Tier-2 reaps deferred because a live
+        # monitor still owns a deploy-staging finalization. Reset on restart (safe:
+        # startup requeue_running_jobs covers the restart path).
+        self.finalizer_defers_total: int = 0
 
     # -- A: zombie-job reaping --------------------------------------------
     def reap_once(self) -> None:
@@ -199,13 +211,34 @@ class JobReaper:
             finished_age = job.get("finished_age_s")
             grace = int(settings.reaper_finalize_grace_s)
             if finished_age is not None and int(finished_age) >= grace:
-                self._reap_known_outcome(job, int(exit_code))
-                return
-            logger.info(
-                "reaper: job %s exit_code=%s recorded %ss ago (< grace %ss) — "
-                "deferring (monitor may still be finalizing)",
-                job_id, exit_code, finished_age, grace,
-            )
+                # ORCH-113 (adr-0043 / D3): even past the grace, a LIVE monitor may
+                # still be running the minutes-long deploy-staging edge sub-gates
+                # in-thread — finished_age is measured from the START of finalization
+                # (the finished_at stamp), and on deploy-staging the heavy advance
+                # (security/merge-gate re-test/coverage/image-freshness) runs AFTER
+                # that stamp and BEFORE _finalize_job. If a live finalizer still owns
+                # this job, DEFER the Tier-2 reap (re-running the advance caused the
+                # false rollback in incident ORCH-111) and fall through to the Tier-3
+                # backstop, which IGNORES the marker so a stuck/dead finalizer is
+                # still reaped in bounded time.
+                if self._finalizer_owns(job):
+                    self.finalizer_defers_total += 1
+                    logger.info(
+                        "reaper: job %s (deploy-staging) still owned by a live "
+                        "finalizer %ss past grace — deferring Tier-2 (Tier-3 backstop "
+                        "at %ss still applies)",
+                        job_id, finished_age, settings.reaper_max_running_s,
+                    )
+                    # fall through to the Tier-3 backstop guard below.
+                else:
+                    self._reap_known_outcome(job, int(exit_code))
+                    return
+            else:
+                logger.info(
+                    "reaper: job %s exit_code=%s recorded %ss ago (< grace %ss) — "
+                    "deferring (monitor may still be finalizing)",
+                    job_id, exit_code, finished_age, grace,
+                )
             # fall through to the Tier-3 backstop guard below.
         else:
             # Tier-1: dead pid, only after `reaper_dead_ticks` consecutive dead ticks.
@@ -400,6 +433,33 @@ class JobReaper:
                            job.get("id"), e)
             return None, None, None
 
+    def _finalizer_owns(self, job: dict) -> bool:
+        """ORCH-113 (adr-0043 / D3): True iff a LIVE monitor still owns this job's
+        ``deploy-staging`` finalization, so the Tier-2 reap must be deferred.
+
+        Order matters for the zero-regression contract: the kill-switch is checked
+        FIRST (disabled -> ``False`` with no DB lookup, so the path is byte-for-byte
+        prior); then the stage is scoped to ``deploy-staging`` only (the sole edge
+        whose in-thread finalization runs for minutes — every other stage is left
+        untouched); only then is the process-local ownership marker consulted. Never
+        raises -> ``False`` on any error (conservative: never block reaping when
+        ownership is unknowable, so the Tier-3 backstop is never neutered).
+        """
+        try:
+            if not settings.reaper_finalizer_liveness_enabled:
+                return False
+            _branch, stage, _wid = self._task_meta(job)
+            if stage != "deploy-staging":
+                return False
+            from . import finalizer_liveness
+            return finalizer_liveness.is_active(job.get("id"))
+        except Exception as e:  # noqa: BLE001 - never break the reap tick
+            logger.warning(
+                "reaper: finalizer-liveness check failed for job %s: %s",
+                job.get("id"), e,
+            )
+            return False
+
     def _notify_failed(self, job: dict, reason: str) -> None:
         try:
             from .notifications import send_telegram
@@ -463,6 +523,14 @@ class JobReaper:
 
     def status(self) -> dict:
         """Reaper snapshot for /queue observability (Р-6)."""
+        # ORCH-113 (adr-0043 / D5): expose the defer counter + the current finalizer
+        # ownership set (read-only, never-raise). Additive keys only — existing keys
+        # are unchanged.
+        try:
+            from . import finalizer_liveness
+            _owned = finalizer_liveness.snapshot()
+        except Exception:  # noqa: BLE001 - observability must never break /queue
+            _owned = {"active": 0, "jobs": []}
         return {
             "enabled": settings.reaper_enabled,
             "interval": self.interval_s,
@@ -470,6 +538,9 @@ class JobReaper:
             "reaped_total": self.reaped_total,
             "last_reaped": self.last_reaped,
             "lease_reclaimed_total": self.lease_reclaimed_total,
+            "finalizer_liveness_enabled": settings.reaper_finalizer_liveness_enabled,
+            "finalizer_defers_total": self.finalizer_defers_total,
+            "finalizer_owned": _owned,
         }
 
 

tests/test_orch113_reaper_finalizer_liveness.py

@@ -0,0 +1,386 @@
+"""ORCH-113 (adr-0043): reaper must not re-run deploy-staging finalization while
+the original finalizer is alive — finalizer-liveness ownership tests (TC-01..TC-08).
+
+Covers the bug from incident ORCH-111 (deployer job 1914): on the
+``deploy-staging -> deploy`` edge the live monitor runs the heavy edge sub-gates
+(security/merge-gate re-test/coverage/image-freshness) in-thread for MINUTES AFTER
+stamping ``agent_runs.finished_at`` and BEFORE ``_finalize_job``. Reaper Tier-2
+measures ``finished_age_s`` from ``finished_at``, so past ``reaper_finalize_grace_s``
+it treated the live, long-finalizing monitor as dead and independently re-ran the
+advance -> a second re-test went red -> false rollback ``deploy-staging ->
+development`` while the original finalizer concurrently merged the PR. State diverged.
+
+The reaper never spawns claude / pytest / docker; we drive the DB directly (a
+'running' jobs row + a linked agent_runs exit_code) and the process-local
+``finalizer_liveness`` marker, then assert the reaper's terminal flip / deferral.
+No network, no subprocess — every external is mocked.
+"""
+import os
+import tempfile
+
+import pytest
+
+# Override env before importing app modules (same convention as test_job_reaper.py).
+os.environ["ORCH_DB_PATH"] = os.path.join(
+    tempfile.gettempdir(), "test_orch113_reaper.db"
+)
+os.environ["ORCH_REPOS_DIR"] = tempfile.gettempdir()
+os.environ["ORCH_GITEA_TOKEN"] = "test-token"
+os.environ["ORCH_PLANE_API_TOKEN"] = "test-token"
+
+import src.db as db
+from src.db import init_db, get_db, get_job
+import src.finalizer_liveness as fl
+import src.job_reaper as jr
+from src.job_reaper import JobReaper
+
+
+@pytest.fixture(autouse=True)
+def fresh_db(tmp_path, monkeypatch):
+    dbfile = tmp_path / "reaper113.db"
+    monkeypatch.setattr(db.settings, "db_path", str(dbfile))
+    init_db()
+    # Each test starts with a clean ownership registry (process-local module state).
+    with fl._LOCK:
+        fl._OWNED.clear()
+    # Default: kill-switch ON (the fix is active) unless a test flips it.
+    monkeypatch.setattr(db.settings, "reaper_finalizer_liveness_enabled", True)
+    yield
+    with fl._LOCK:
+        fl._OWNED.clear()
+
+
+# --- helpers ----------------------------------------------------------------
+def _make_task(repo="orchestrator", branch="feature/orch113",
+               stage="deploy-staging", work_item_id="ORCH-113"):
+    conn = get_db()
+    cur = conn.execute(
+        "INSERT INTO tasks (plane_id, work_item_id, repo, branch, stage) "
+        "VALUES (?, ?, ?, ?, ?)",
+        (work_item_id, work_item_id, repo, branch, stage),
+    )
+    tid = cur.lastrowid
+    conn.commit()
+    conn.close()
+    return tid
+
+
+def _make_running_job(agent="deployer", repo="orchestrator", task_id=None,
+                      pid=None, age_s=0, attempts=0, max_attempts=2,
+                      run_id=None, exit_code=0, finished_age_s=600):
+    """Insert a job already in 'running' with a linked agent_runs row.
+
+    ``finished_at`` is back-dated by ``finished_age_s`` so the Tier-2 grace
+    (default 300) is satisfied by default; pass a small value to stay within grace.
+    """
+    conn = get_db()
+    if run_id is None and exit_code is not None:
+        cur = conn.execute(
+            "INSERT INTO agent_runs (task_id, agent, finished_at, exit_code) "
+            "VALUES (?, ?, datetime('now', ?), ?)",
+            (task_id, agent, f"-{int(finished_age_s)} seconds", exit_code),
+        )
+        run_id = cur.lastrowid
+    cur = conn.execute(
+        "INSERT INTO jobs (agent, repo, task_id, status, attempts, max_attempts, "
+        "run_id, pid, started_at) "
+        "VALUES (?, ?, ?, 'running', ?, ?, ?, ?, datetime('now', ?))",
+        (agent, repo, task_id, attempts, max_attempts, run_id, pid,
+         f"-{int(age_s)} seconds"),
+    )
+    job_id = cur.lastrowid
+    conn.commit()
+    conn.close()
+    return job_id
+
+
+def _spy_advance(monkeypatch, side_effect=None):
+    """Patch launcher._try_advance_stage with a call recorder.
+
+    Returns a mutable ``calls`` list. ``side_effect(run_id, agent, repo, branch)``
+    runs on each call (e.g. to simulate the false rollback to development).
+    """
+    import src.agents.launcher as L
+    calls: list = []
+
+    def _fake(run_id, agent, repo, branch):
+        calls.append((run_id, agent, repo, branch))
+        if side_effect is not None:
+            side_effect(run_id, agent, repo, branch)
+
+    monkeypatch.setattr(L.launcher, "_try_advance_stage", _fake)
+    return calls
+
+
+def _green_gate(monkeypatch):
+    """Force the read-only canonical-QG pre-eval green (staging SUCCESS)."""
+    monkeypatch.setattr(JobReaper, "_gate_is_green",
+                        lambda self, stage, job, branch, wid: True)
+
+
+# --- TC-01: live finalizer on deploy-staging is NOT reaped (AC-1/FR-1) ------
+def test_tc01_live_finalizer_deploy_staging_not_reaped(monkeypatch):
+    """exit_code=0 and finished_age_s >= grace, but the finalizer is ALIVE (marked)
+    -> reaper does NOT call _try_advance_stage; the row stays running; defer logged."""
+    _green_gate(monkeypatch)
+    calls = _spy_advance(monkeypatch)
+    tid = _make_task(stage="deploy-staging")
+    jid = _make_running_job(task_id=tid, exit_code=0, finished_age_s=600)
+    # A live monitor owns this finalization.
+    fl.mark(jid, run_id=1, stage="deploy-staging")
+
+    r = JobReaper()
+    r.reap_once()
+
+    assert get_job(jid)["status"] == "running"   # not reaped
+    assert calls == []                            # no second advance
+    assert r.finalizer_defers_total == 1
+    assert r.reaped_total == 0
+
+
+# --- TC-02: strict ownership — non-owner runs zero side effects (AC-2/FR-2) --
+def test_tc02_non_owner_runs_no_edge_gates(monkeypatch):
+    """While a live finalizer owns the (job, stage), a racing reaper tick performs
+    NO side-effectful advance/merge-gate/re-test (zero side effects)."""
+    _green_gate(monkeypatch)
+    calls = _spy_advance(monkeypatch)
+    tid = _make_task(stage="deploy-staging")
+    jid = _make_running_job(task_id=tid, exit_code=0, finished_age_s=900)
+    fl.mark(jid, run_id=7, stage="deploy-staging")
+
+    r = JobReaper()
+    # Several ticks while ownership is held — still zero advances, still running.
+    for _ in range(3):
+        r.reap_once()
+
+    assert calls == []
+    assert get_job(jid)["status"] == "running"
+    assert r.finalizer_defers_total == 3
+
+
+# --- TC-03: a genuinely dead finalizer is still reaped (AC-3/FR-4) ----------
+def test_tc03_dead_finalizer_still_reaped_tier2(monkeypatch):
+    """No ownership marker (finalizer dead) -> reaper reaps via Tier-2 as before."""
+    _green_gate(monkeypatch)
+    calls = _spy_advance(monkeypatch)
+    tid = _make_task(stage="deploy-staging")
+    jid = _make_running_job(task_id=tid, exit_code=0, finished_age_s=600)
+    # No fl.mark() -> ownership absent (finalizer dead).
+
+    r = JobReaper()
+    r.reap_once()
+
+    assert get_job(jid)["status"] == "done"   # reaped to done (gate green)
+    assert len(calls) == 1                     # advance driven exactly once
+    assert r.finalizer_defers_total == 0
+
+
+def test_tc03_tier3_backstop_ignores_marker(monkeypatch):
+    """Even with an active ownership marker, a job past reaper_max_running_s is
+    reaped by the Tier-3 backstop (a stuck finalizer is never immortal)."""
+    monkeypatch.setattr(db.settings, "reaper_max_running_s", 1000)
+    tid = _make_task(stage="deploy-staging")
+    # age beyond the backstop ceiling; exit_code=0 within grace so Tier-2 defers,
+    # but Tier-3 must still fire.
+    jid = _make_running_job(task_id=tid, exit_code=0, finished_age_s=10,
+                            age_s=2000, attempts=0, max_attempts=2)
+    fl.mark(jid, run_id=3, stage="deploy-staging")
+
+    r = JobReaper()
+    r.reap_once()
+
+    # Backstop reaps to a retry (attempts<max) -> queued, regardless of the marker.
+    assert get_job(jid)["status"] == "queued"
+    assert r.reaped_total == 1
+
+
+# --- TC-04: idempotency under race — exactly-once advance (AC-2/AC-4/FR-2) ---
+def test_tc04_idempotent_no_second_advance_under_race(monkeypatch):
+    """Monitor finalizing (owns the job) + concurrent reaper ticks -> the heavy
+    edge-gate advance is NEVER duplicated by the reaper; no false rollback."""
+    rolled_back = {"hit": False}
+
+    def _rollback(run_id, agent, repo, branch):
+        # Simulate the incident: a second advance rolls back to development.
+        conn = get_db()
+        conn.execute("UPDATE tasks SET stage='development' WHERE branch=?", (branch,))
+        conn.commit()
+        conn.close()
+        rolled_back["hit"] = True
+
+    _green_gate(monkeypatch)
+    calls = _spy_advance(monkeypatch, side_effect=_rollback)
+    tid = _make_task(stage="deploy-staging")
+    jid = _make_running_job(task_id=tid, exit_code=0, finished_age_s=1200)
+    fl.mark(jid, run_id=9, stage="deploy-staging")
+
+    r = JobReaper()
+    for _ in range(5):
+        r.reap_once()
+
+    assert calls == []                # reaper never re-ran the advance
+    assert rolled_back["hit"] is False
+    # The task is NOT rolled back; the live finalizer remains the sole driver.
+    conn = get_db()
+    stage = conn.execute("SELECT stage FROM tasks WHERE id=?", (tid,)).fetchone()[0]
+    conn.close()
+    assert stage == "deploy-staging"
+
+
+# --- TC-05: MANDATORY regression for incident ORCH-111 (AC-4/FR-5) ----------
+def test_tc05_orch111_no_false_rollback_no_retry_increment(monkeypatch):
+    """Long (> grace) deploy-staging finalization at staging_status=SUCCESS while
+    the deploy finalizer concurrently reaches success/merge -> reaper does NOT roll
+    back deploy-staging -> development and does NOT increment developer-retry; the
+    task keeps a single consistent state. RED before the fix, GREEN after."""
+    def _rollback(run_id, agent, repo, branch):
+        # Simulate the incident: a second advance rolls the task back to development
+        # and spawns a fresh developer run (the developer-retry count is derived from
+        # agent_runs — stage_engine.developer_retry_count).
+        conn = get_db()
+        conn.execute("UPDATE tasks SET stage='development' WHERE branch=?", (branch,))
+        _t = conn.execute("SELECT id FROM tasks WHERE branch=?", (branch,)).fetchone()
+        conn.execute(
+            "INSERT INTO agent_runs (task_id, agent) VALUES (?, 'developer')",
+            (_t[0],),
+        )
+        conn.commit()
+        conn.close()
+
+    from src.stage_engine import developer_retry_count
+    _green_gate(monkeypatch)                       # staging_status SUCCESS
+    calls = _spy_advance(monkeypatch, side_effect=_rollback)
+    tid = _make_task(stage="deploy-staging")
+    jid = _make_running_job(task_id=tid, exit_code=0, finished_age_s=1500)
+    # The original finalizer is still alive (running the heavy edge sub-gates).
+    fl.mark(jid, run_id=11, stage="deploy-staging")
+
+    r = JobReaper()
+    r.reap_once()
+
+    # No second advance => no false rollback, no developer-retry increment.
+    assert calls == []
+    conn = get_db()
+    stage = conn.execute("SELECT stage FROM tasks WHERE id=?", (tid,)).fetchone()[0]
+    conn.close()
+    assert stage == "deploy-staging"            # NOT rolled back to development
+    assert developer_retry_count(tid) == 0       # developer-retry NOT incremented
+    assert get_job(jid)["status"] == "running"
+    assert r.finalizer_defers_total == 1
+
+
+# --- TC-06: compatibility guard — kill-switch / non-deploy-staging (AC-5) ----
+def test_tc06_killswitch_off_byte_for_byte_prior(monkeypatch):
+    """Kill-switch OFF -> the marker is ignored; a deploy-staging exit0/past-grace
+    job is reaped exactly as before the fix (advance driven once)."""
+    monkeypatch.setattr(db.settings, "reaper_finalizer_liveness_enabled", False)
+    _green_gate(monkeypatch)
+    calls = _spy_advance(monkeypatch)
+    tid = _make_task(stage="deploy-staging")
+    jid = _make_running_job(task_id=tid, exit_code=0, finished_age_s=600)
+    fl.mark(jid, run_id=5, stage="deploy-staging")   # marked, but ignored
+
+    r = JobReaper()
+    r.reap_once()
+
+    assert get_job(jid)["status"] == "done"
+    assert len(calls) == 1
+    assert r.finalizer_defers_total == 0
+
+
+def test_tc06_non_deploy_staging_stage_not_consulted(monkeypatch):
+    """A non-deploy-staging stage is never consulted -> reaped as before even when
+    an (irrelevant) marker happens to be present."""
+    _green_gate(monkeypatch)
+    calls = _spy_advance(monkeypatch)
+    tid = _make_task(stage="testing")     # deployer also owns 'testing'
+    jid = _make_running_job(task_id=tid, agent="deployer", exit_code=0,
+                            finished_age_s=600)
+    fl.mark(jid, run_id=6, stage="testing")
+
+    r = JobReaper()
+    r.reap_once()
+
+    assert get_job(jid)["status"] == "done"
+    assert len(calls) == 1
+    assert r.finalizer_defers_total == 0
+
+
+def test_tc06_within_grace_unchanged(monkeypatch):
+    """Within the finalization grace the Tier-2 path is unchanged (deferred, not
+    reaped) regardless of the marker — the fix only acts PAST the grace."""
+    monkeypatch.setattr(db.settings, "reaper_finalize_grace_s", 300)
+    _green_gate(monkeypatch)
+    calls = _spy_advance(monkeypatch)
+    tid = _make_task(stage="deploy-staging")
+    jid = _make_running_job(task_id=tid, exit_code=0, finished_age_s=5)  # < grace
+
+    r = JobReaper()
+    r.reap_once()
+
+    assert get_job(jid)["status"] == "running"
+    assert calls == []
+    # Within-grace deferral is the legacy path, not a finalizer-liveness defer.
+    assert r.finalizer_defers_total == 0
+
+
+# --- TC-07: cross-cutting budget invariant (NFR-6/AC-5) ---------------------
+def test_tc07_budget_invariant_preserved():
+    """reaper_max_running_s (5400) > Σ(deploy-staging gate-work) + grace; the fix
+    changed neither the grace nor the ceiling (ORCH-065/109/110 invariant)."""
+    s = jr.settings
+    # The fix did not touch these (zero schema/budget change).
+    assert s.reaper_finalize_grace_s == 300
+    assert s.reaper_max_running_s == 5400
+    # Conservative Σ of the heavy deploy-staging gate-work + grace must fit.
+    sigma = s.merge_retest_timeout_s + s.coverage_run_timeout_s
+    assert s.reaper_max_running_s > sigma + s.reaper_finalize_grace_s
+
+
+# --- TC-08: never-raise — a fault in the liveness path degrades safely -------
+def test_tc08_liveness_error_never_breaks_tick(monkeypatch):
+    """An exception inside the ownership consultation must not crash the tick; the
+    job is still processed (conservatively reaped, never blocked)."""
+    def _boom(job_id):
+        raise RuntimeError("registry exploded")
+
+    monkeypatch.setattr(fl, "is_active", _boom)
+    _green_gate(monkeypatch)
+    calls = _spy_advance(monkeypatch)
+    tid = _make_task(stage="deploy-staging")
+    jid = _make_running_job(task_id=tid, exit_code=0, finished_age_s=600)
+    fl.mark(jid, run_id=2, stage="deploy-staging")
+
+    r = JobReaper()
+    r.reap_once()  # must not raise
+
+    # is_active raised -> _finalizer_owns conservatively returns False -> reaped.
+    assert get_job(jid)["status"] == "done"
+    assert len(calls) == 1
+
+
+def test_tc08_reap_once_isolates_and_never_raises(monkeypatch):
+    """A fault while resolving one job's metadata is isolated; reap_once never
+    raises and other jobs are still processed."""
+    def _boom(self, job):
+        raise RuntimeError("meta exploded")
+
+    monkeypatch.setattr(JobReaper, "_task_meta", _boom)
+    tid = _make_task(stage="deploy-staging")
+    _make_running_job(task_id=tid, exit_code=0, finished_age_s=600)
+
+    r = JobReaper()
+    r.reap_once()  # outer + per-job never-raise -> no exception propagates
+
+
+def test_tc08_finalizer_liveness_leaf_never_raises():
+    """The leaf degrades safely on bad input / None job_id."""
+    fl.mark(None, None, None)          # no-op, no raise
+    fl.clear(None)                     # no-op, no raise
+    assert fl.is_active(None) is False
+    fl.mark(1234, 1, "deploy-staging")
+    assert fl.is_active(1234) is True
+    snap = fl.snapshot()
+    assert snap["active"] >= 1 and 1234 in snap["jobs"]
+    fl.clear(1234)
+    assert fl.is_active(1234) is False