"""Unified stage engine (ORCH-4 / M-3).

Single source of truth for "an agent finished / a human approved -> run the
stage's quality gate and either advance the pipeline or roll it back".

Before ORCH-4 this logic was duplicated in two places that had silently
diverged:
  - src/agents/launcher.py::_try_advance_stage (sync, rich business logic:
    analyst approved-flow, reviewer REQUEST_CHANGES rollback+retry, tester FAIL
    rollback+retry, architect conflict rollback) — but it picked the next agent
    with get_agent_for_stage(next_stage), which is WRONG.
  - src/webhooks/plane.py::_try_advance_stage (async, leaner, but it had the
    check_review_approved PR-by-branch dispatch and used the CORRECT
    get_agent_for_stage(current_stage)).

This module merges both into one sync `advance_stage(...)`. launcher calls it
directly; the plane webhook calls it through asyncio.to_thread so there is
exactly one implementation.

Agent-selection bug fix (ORCH-4):
  stages.py defines `agent` as "the agent to launch when advancing FROM this
  stage". So when advancing current -> next, the correct agent to launch is
  get_agent_for_stage(current_stage). launcher's old next_stage lookup skipped a
  stage (e.g. analysis->architecture launched 'developer' instead of
  'architect'). plane and gitea already used current_stage; we unify on that.
"""

import logging
import os
import time
from dataclasses import dataclass, field

from .db import get_db, update_task_stage, enqueue_job
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 .security_gate import extract_security_findings
from .qg.checks import QG_CHECKS
from . import merge_gate
from . import self_deploy
from . import post_deploy
from .notifications import (
    notify_stage_change,
    notify_qg_failure,
    notify_approve_requested,
    send_telegram,
    link_for,
)
from .plane_sync import (
    notify_stage_change as plane_notify_stage,
    notify_qg_failure as plane_notify_qg,
    add_comment as plane_add_comment,
    set_issue_in_review,
    set_issue_needs_input,
    set_issue_in_progress,
    set_issue_blocked,
    set_issue_done,
    set_issue_analysis,
    set_issue_awaiting_deploy,
    set_issue_deploying,
    set_issue_monitoring,
)
from .config import settings

logger = logging.getLogger("orchestrator.stage_engine")

MAX_DEVELOPER_RETRIES = 3


@dataclass
class AdvanceResult:
    """Outcome of an advance_stage() call (mostly for tests/observability)."""

    advanced: bool = False
    from_stage: str | None = None
    to_stage: str | None = None
    enqueued_agent: str | None = None
    enqueued_job_id: int | None = None
    qg_name: str | None = None
    qg_passed: bool | None = None
    qg_reason: str | None = None
    rolled_back_to: str | None = None
    alerted: bool = False
    note: str | None = None
    notes: list = field(default_factory=list)


def _run_qg(qg_name: str, repo: str, work_item_id: str, branch: str):
    """Dispatch a quality-gate check to the right signature and run it.

    Signatures (unified from launcher + plane):
      - check_ci_green / check_tests_local      -> (repo, branch)
      - check_review_approved                   -> (repo, pr_number) [PR found by branch]
      - everything else (artifact checks)       -> (repo, work_item_id, branch)

    Returns (passed: bool, reason: str).
    """
    check_fn = QG_CHECKS.get(qg_name)
    if not check_fn:
        logger.error(f"QG function '{qg_name}' not found in registry")
        return False, f"Unknown QG: {qg_name}"

    if qg_name in ("check_ci_green", "check_tests_local"):
        # (repo, branch) — already worktree-aware.
        return check_fn(repo, branch)

    if qg_name == "check_review_approved":
        # Special case kept from plane: find the open PR for this branch via
        # Gitea, then check it; fall back to a file-based review marker.
        return _check_review_approved_by_branch(check_fn, repo, work_item_id, branch)

    # All other artifact checks: (repo, work_item_id, branch). Pass branch so the
    # check reads from the task worktree (ORCH-2 / S-4).
    return check_fn(repo, work_item_id or "", branch)


def _check_review_approved_by_branch(check_fn, repo: str, work_item_id: str, branch: str):
    """check_review_approved dispatch preserved from plane._try_advance_stage.

    Finds the open PR whose head ref == branch via the Gitea API and runs
    check_review_approved(repo, pr_number). If no open PR exists, falls back to a
    file-based review marker (12-review.md / 09-review.md) like the original.
    """
    import httpx as _httpx

    owner = settings.gitea_owner
    url = f"{settings.gitea_url}/api/v1/repos/{owner}/{repo}/pulls?state=open&limit=50"
    headers = {"Authorization": f"token {settings.gitea_token}"}
    try:
        resp = _httpx.get(url, headers=headers, timeout=10)
        prs = resp.json()
        pr_number = None
        for pr in prs:
            if pr.get("head", {}).get("ref") == branch:
                pr_number = pr["number"]
                break
        if pr_number:
            return check_fn(repo, pr_number)
        # No open PR but a review file may exist — check file-based.
        wt = get_worktree_path(repo, branch)
        if not os.path.isdir(wt):
            wt = os.path.join(settings.repos_dir, repo)
        review_path = os.path.join(wt, f"docs/work-items/{work_item_id}/12-review.md")
        review_path2 = os.path.join(wt, f"docs/work-items/{work_item_id}/09-review.md")
        if os.path.isfile(review_path) or os.path.isfile(review_path2):
            return True, "Review file exists (file-based approval)"
        return False, "No open PR found and no review file"
    except Exception as e:
        return False, f"Error finding PR: {e}"


def developer_retry_count(task_id: int) -> int:
    """How many developer runs have already happened for this task.

    Single source of truth for the developer-retry count: the rollback path
    (REQUEST_CHANGES / test-fail / merge-gate) and the ORCH-060 reconciler guard
    both read the cap from here, so the SQL is never duplicated. ``task`` is
    considered *escalated* once this reaches ``MAX_DEVELOPER_RETRIES``.
    """
    conn = get_db()
    n = conn.execute(
        "SELECT COUNT(*) FROM agent_runs WHERE task_id=? AND agent='developer'",
        (task_id,),
    ).fetchone()[0]
    conn.close()
    return n


# Backward-compat private alias — existing internal call sites keep working.
_developer_retry_count = developer_retry_count


def advance_stage(
    task_id: int,
    current_stage: str,
    repo: str,
    work_item_id: str,
    branch: str,
    finished_agent: str | None = None,
    *,
    confirm_deploy: bool = False,
) -> AdvanceResult:
    """Run the current stage's quality gate and advance / roll back the pipeline.

    This is the single merged implementation (ORCH-4 / M-3). It is synchronous;
    the async plane webhook calls it via asyncio.to_thread.

    Args:
      task_id:        tasks.id
      current_stage:  the stage the task is currently in
      repo:           repository name
      work_item_id:   Plane work item id (may be "" / None)
      branch:         feature branch
      finished_agent: the agent that just finished (launcher path). Drives the
                      approved/REQUEST_CHANGES/tester/architect branches. In the
                      plane webhook path it is None, so those agent-specific
                      branches simply do not trigger (matches old plane behavior).
      confirm_deploy: ORCH-059 — keyword-only signal that the human flipped the
                      issue to the dedicated "Confirm Deploy" status. ONLY this
                      signal initiates Phase B of the self-hosting prod deploy on
                      the `deploy` stage. A plain `Approved` on `deploy`
                      (confirm_deploy=False) is a deliberate no-op (no prod
                      deploy, no false БАГ-8 rollback). All non-webhook callers
                      leave it at the default.

    Returns AdvanceResult describing what happened.
    """
    result = AdvanceResult(from_stage=current_stage)
    agent = finished_agent
    try:
        qg_name = get_qg_for_stage(current_stage)
        next_stage = get_next_stage(current_stage)
        result.qg_name = qg_name
        result.to_stage = next_stage

        if not next_stage:
            logger.info(f"Task {task_id}: already at terminal stage '{current_stage}'")
            result.note = "terminal"
            return result

        # --- ORCH-036/059 Phase B: "Confirm Deploy" on `deploy` -> initiate ----
        # ORCH-059: the prod-deploy trigger is now the DEDICATED "Confirm Deploy"
        # status (confirm_deploy=True), NOT the overloaded "Approved". On the
        # `deploy` stage (finished_agent is None) for the self-hosting repo we
        # always return early WITHOUT running check_deploy_status (the verdict
        # does not exist yet — running the gate now would read a stale/absent log
        # and falsely roll back, R-2/БАГ-8), but we only initiate the DETACHED
        # host deploy + enqueue the finalizer when confirm_deploy is set. A plain
        # Approved (confirm_deploy=False) is a deliberate no-op — it neither
        # deploys nor rolls back (TRZ-3/AC-3). The finalizer (Phase C,
        # finished_agent="deployer") records the verdict later; that path is NOT
        # intercepted here (it requires finished_agent set).
        if (
            current_stage == "deploy"
            and finished_agent is None
            and settings.deploy_require_manual_approve
            and self_deploy.self_deploy_applies(repo)
        ):
            if confirm_deploy:
                _handle_self_deploy_phase_b(task_id, repo, work_item_id, branch, result)
            else:
                result.note = "approved-on-deploy-noop"
                logger.info(
                    f"Task {task_id}: Approved on `deploy` without Confirm Deploy "
                    f"— no-op (prod deploy requires the 'Confirm Deploy' status)"
                )
            return result

        # --- Quality gate ----------------------------------------------------
        if qg_name and qg_name in QG_CHECKS:
            # Human-approval gate: split by path.
            if qg_name == "check_analysis_approved":
                # Launcher path (analyst just finished): set In Review + ask for
                # the Approved status. This gate never advances on its own -- a
                # human Approved verdict does that.
                if agent == "analyst":
                    _handle_analysis_approved_flow(
                        task_id, current_stage, repo, work_item_id, branch, agent, result
                    )
                    return result
                # Webhook Approved-verdict path (agent is None): the human flipped
                # the Plane status to Approved, which IS the approval. The gate is
                # satisfied -- do NOT re-run check_analysis_approved (it looks for
                # an :approved: *comment* and would block on a status-only
                # approval). Mark it passed and fall through to the Advance block.
                result.qg_name = qg_name
                result.qg_passed = True
                result.qg_reason = "approved-via-status"
            else:
                passed, reason = _run_qg(qg_name, repo, work_item_id, branch)
                result.qg_passed = passed
                result.qg_reason = reason

                if not passed:
                    logger.info(
                        f"Task {task_id}: QG '{qg_name}' not passed after {agent}: {reason}"
                    )
                    # Behaviour parity:
                    #  - webhook path (finished_agent is None): emit the generic
                    #    QG-failure notification, exactly like the old plane handler.
                    #  - launcher path (finished_agent set): NO generic notification;
                    #    the rollback branches below own their own messaging, exactly
                    #    like the old launcher handler.
                    if agent is None:
                        notify_qg_failure(task_id, current_stage, qg_name, reason)
                        plane_notify_qg(work_item_id, current_stage, qg_name, reason)

                    _handle_qg_failure_rollbacks(
                        task_id, current_stage, repo, work_item_id, branch,
                        agent, qg_name, reason, result,
                    )
                    return result

        elif qg_name:
            # QG name set but not registered — do not advance (launcher behavior).
            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":
            # --- ORCH-022 security sub-gate (deploy-staging -> deploy edge) -----
            # Run FIRST among the edge sub-gates (BEFORE the merge-gate and the
            # image-freshness rebuild): it is cheap (read-only scan) and we want to
            # fail BEFORE the expensive rebase/rebuild (07-infra I-6). Deterministic:
            # gitleaks (offline secret-scan) + pip-audit (CVE audit). FAIL -> rollback
            # to development + developer-retry (cap MAX_DEVELOPER_RETRIES). It owns
            # the outcome on intervention (mirrors the merge-gate / image-freshness).
            if _handle_security_gate(
                task_id, current_stage, repo, work_item_id, branch, agent, result
            ):
                return result

            if _handle_merge_gate(
                task_id, current_stage, repo, work_item_id, branch, agent, result
            ):
                return result

            # --- ORCH-058 freshness sub-gate (deploy-staging -> deploy edge) ---
            # AFTER the merge-gate finalised the validated HEAD and BEFORE Phase A.
            # Rebuilds the staging image from that validated commit + recreates 8501
            # so the artefact we validate is the exact one promoted to prod (AC-4).
            # FAIL -> rollback to development (mirrors the merge-gate). Like the
            # merge-gate it owns the outcome on intervention.
            if _handle_image_freshness(
                task_id, current_stage, repo, work_item_id, branch, agent, result
            ):
                return result

        # --- ORCH-036 Phase A: request approve before the prod deploy ---------
        # On the deploy-staging -> deploy edge, AFTER a green check_staging_status
        # and the merge-gate, the self-hosting repo does NOT auto-launch a prod
        # deployer. Instead advance the STAGE to `deploy`, put the issue into an
        # approval-pending state and wait for a human Approved (Phase B). The
        # merge lease stays HELD across the wait (released on done / rollback).
        if (
            current_stage == "deploy-staging"
            and settings.deploy_require_manual_approve
            and self_deploy.self_deploy_applies(repo)
        ):
            _handle_self_deploy_phase_a(
                task_id, current_stage, repo, work_item_id, branch, result
            )
            return result

        # --- ORCH-071 merge-verify under-gate (deploy -> done edge) ----------
        # The SINGLE choke-point that gates EVERY path into terminal `done`
        # (finalizer Phase C, reconciler F-1, job-reaper re-drive) on a CONFIRMED
        # merge of the feature PR into `main`. For the self-hosting repo the
        # deterministic self-deploy path never runs the LLM `deployer` that used to
        # merge the PR, so a green deploy could reach `done` while the PR stayed
        # `open` (phantom merge, ORCH-071). This врезка runs a deterministic
        # merge-actor + post-deploy verification BEFORE update_task_stage; if the
        # merge is not confirmed it HOLDs (alert, NO done, NO rollback) and returns
        # without advancing. Not a STAGE_TRANSITIONS edge / registered QG — it is an
        # edge sub-gate (mirrors the merge-gate врезка), so those contracts are
        # unchanged. No-op for non-self repos / kill-switch off (1:1 prior behaviour).
        if current_stage == "deploy" and next_stage == "done":
            if _handle_merge_verify(task_id, repo, work_item_id, branch, result):
                return result

        # --- Advance ---------------------------------------------------------
        update_task_stage(task_id, next_stage)
        # Telegram live tracker: the analysis->architecture advance is the human
        # Approved gate clearing -> stamp the END of "Ревью БРД" (the only
        # human time). Idempotent: only the first stamp counts.
        if current_stage == "analysis" and next_stage == "architecture":
            try:
                from .db import mark_brd_review_ended
                mark_brd_review_ended(task_id)
            except Exception as e:
                logger.warning(f"Task {task_id}: brd review end stamp failed: {e}")
        notify_stage_change(task_id, current_stage, next_stage)
        plane_notify_stage(work_item_id, current_stage, next_stage)
        result.advanced = True
        logger.info(
            f"Task {task_id}: {current_stage} -> {next_stage} "
            f"(auto-advance after {agent})"
        )

        # --- Terminal sync: deploy -> done must reach Plane's Done -----------
        # When the deployer's check_deploy_status passes we advance to the
        # terminal 'done' stage. Previously a merged-PR webhook completed the
        # task out-of-band and Plane stuck on In Progress. Now done flows through
        # here, so explicitly drive the Plane issue into the terminal Done state
        # (PLANE_STATES['done'] — mapping unchanged) in addition to the
        # stage-change comment above.
        # ORCH-066 (AC-8/AC-9): split terminal-sync by whether post-deploy
        # monitoring applies. For self-hosting (post_deploy_applies==True) the
        # task enters a `Monitoring after Deploy` window, NOT terminal Done yet —
        # the monitor finalises Done/Blocked (run_post_deploy_monitor). For
        # non-self repos the behaviour is unchanged: terminal Done immediately.
        # Where the `Monitoring after Deploy` status is absent, set_issue_monitoring
        # degrades to the project's Done UUID -> identical to today.
        if next_stage == "done" and work_item_id:
            try:
                if post_deploy.post_deploy_applies(repo):
                    set_issue_monitoring(work_item_id)
                    logger.info(
                        f"Task {task_id}: deploy->done (self), Plane state -> "
                        f"Monitoring after Deploy (post-deploy window)"
                    )
                else:
                    set_issue_done(work_item_id)
                    logger.info(
                        f"Task {task_id}: deploy->done, Plane state forced to Done"
                    )
            except Exception as e:
                logger.error(f"Task {task_id}: failed to set Plane terminal state: {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}")

        # ORCH-021: arm post-deploy monitoring PAST `done`. Responsibility extends
        # beyond the restart-time health-check to catch the "green deploy, red prod"
        # class (ET-8). Idempotent (sentinel `armed`) + conditional (applies()), so a
        # double webhook / reconciler / finalizer re-driving `done` never doubles it
        # and non-applicable repos are untouched. never-raise (arm_monitor + guard).
        if next_stage == "done" and post_deploy.post_deploy_applies(repo):
            try:
                post_deploy.arm_monitor(repo, work_item_id, branch, task_id)
            except Exception as e:  # noqa: BLE001 - monitoring must never crash done
                logger.warning(f"Task {task_id}: post-deploy arm failed: {e}")

        # --- Launch the next agent (ORCH-4 fix: current_stage, not next) -----
        next_agent = get_agent_for_stage(current_stage)
        if next_agent:
            task_desc = (
                f"Work item: {work_item_id}\nRepo: {repo}\n"
                f"Branch: {branch}\nStage: {next_stage}"
            )
            new_job_id = enqueue_job(next_agent, repo, task_desc, task_id=task_id)
            result.enqueued_agent = next_agent
            result.enqueued_job_id = new_job_id
            logger.info(
                f"Task {task_id}: enqueued '{next_agent}' (job_id={new_job_id})"
            )

        return result

    except Exception as e:
        logger.error(f"advance_stage failed for task_id={task_id}: {e}")
        result.note = f"error: {e}"
        return result


def advance_if_gate_passed(
    task_id: int,
    current_stage: str,
    repo: str,
    work_item_id: str,
    branch: str,
) -> AdvanceResult | None:
    """ORCH-053 (F-1): reconcile a stuck stage by advancing it ONLY if its
    quality gate is already green — without spamming failure notifications.

    This is the thin wrapper the reconciler uses so that:

      * The source of truth stays the GATE, and the advance path stays the
        UNCHANGED unified ``advance_stage(..., finished_agent=None)`` (the same
        path the Plane Approved-webhook uses). The reconciler never duplicates
        ``update_task_stage`` / ``enqueue_job`` (AC-2).

      * On a stable-RED gate the sweeper is structurally silent: we do a cheap
        read-only pre-evaluation of the gate and, if it fails, return ``None``
        WITHOUT ever calling ``advance_stage`` — so the QG-failure notification
        branch inside ``advance_stage`` (``agent is None`` ->
        ``notify_qg_failure`` + ``plane_notify_qg``) cannot fire on any tick
        (AC-9). Spam is impossible by construction.

    ``analysis`` is intentionally NOT reconciled here: its gate
    (``check_analysis_approved``) is a HUMAN gate; with ``finished_agent=None``
    ``advance_stage`` would treat it as approved-via-status and could advance an
    unapproved BRD. The analysis advance is owned by the Plane-side reconciler
    (F-2), which checks the real Plane status (AC-16).

    Returns the ``AdvanceResult`` from ``advance_stage`` when the gate passed,
    or ``None`` when the stage is not eligible / the gate is red / on any error
    (never raises — the caller isolates per-task failures).
    """
    try:
        # AC-16: F-1 never reconciles the human analysis gate.
        if current_stage == "analysis":
            return None

        qg_name = get_qg_for_stage(current_stage)
        if not qg_name:
            # created / done -> no gate to evaluate.
            return None

        # Read-only pre-evaluation with the SAME dispatcher the webhook path uses.
        passed, reason = _run_qg(qg_name, repo, work_item_id, branch)
        if not passed:
            # Stable-red -> stay silent (no advance_stage call -> no QG-failure
            # notification on this or any later tick).
            logger.debug(
                f"reconciler: task {task_id} gate '{qg_name}' still red "
                f"({reason}); leaving on '{current_stage}'"
            )
            return None

        # Gate is green: advance via the unchanged unified path. It re-runs the
        # (idempotent, read-only) gate, advances the stage, sends the STANDARD
        # advance notifications and enqueues the next agent.
        return advance_stage(
            task_id, current_stage, repo, work_item_id, branch, finished_agent=None
        )
    except Exception as e:  # noqa: BLE001 - never-raise per ORCH-053 NFR
        logger.error(
            f"advance_if_gate_passed failed for task_id={task_id} "
            f"stage={current_stage}: {e}"
        )
        return None


def _build_analyst_ready_comment(
    repo: str, work_item_id: str, branch: str, task_id: int | None = None
) -> str:
    """ORCH-016: analyst "artifacts ready" comment via the unified status helper.

    Historically this function hand-built the HTML for the analyst's BUG-C
    status-only verdict comment (PR #12 / #13). After ORCH-016 / ADR-001 \u00a71 every
    agent goes through the single ``usage.build_status_comment(...)`` hot path,
    so this is now a thin compatibility wrapper that:

      - keeps the same 3-positional signature that ``_handle_analysis_approved_flow``
        and the regression tests (``tests/test_analyst_comment.py``) already call,
      - adds an optional ``task_id`` so the duration line for the analyst can be
        resolved via the DB fallback (AC-14: analyst's ``_duration_s`` isn't in
        scope of stage_engine, hence the fallback),
      - locates the worktree so AC-8 graceful skipping of missing analyst
        artifacts and ``gitea_public_url`` clickability work exactly as before.

    All historical text contracts are preserved by the analyst branch inside
    ``build_status_comment``: \u00abApproved\u00bb, \u00abRejected\u00bb, no \u00ab:approved:\u00bb, no
    \u00abIn Progress\u00bb \u2014 the existing test_analyst_comment.py assertions still hold.
    """
    from .usage import build_status_comment
    try:
        worktree_root = get_worktree_path(repo, branch)
    except Exception:
        worktree_root = None
    return build_status_comment(
        "analyst",
        repo=repo,
        branch=branch,
        work_item_id=work_item_id,
        task_id=task_id,
        worktree_root=worktree_root,
    )


def _handle_analysis_approved_flow(
    task_id, current_stage, repo, work_item_id, branch, agent, result: AdvanceResult
):
    """Analyst approved-flow (launcher only).

    Only triggers when the analyst just finished (agent == 'analyst') in the
    launcher path. Decides between: artifacts ready -> In Review + request
    :approved:; questions file -> Needs Input; otherwise a warning comment.
    This gate never advances on its own (human approval does that via the plane
    webhook), matching the original launcher behavior.
    """
    result.qg_name = "check_analysis_approved"
    result.note = "analysis-approval-gate"
    if not (agent == "analyst" and work_item_id):
        return

    files_check = QG_CHECKS.get("check_analysis_complete")
    if not files_check:
        return

    files_ok, _ = files_check(repo, work_item_id, branch)
    if files_ok:
        # Full artifacts ready -> In Review, ask for the Approved STATUS (BUG C).
        set_issue_in_review(work_item_id)
        plane_add_comment(
            work_item_id,
            # task_id is threaded through so build_status_comment can resolve the
            # analyst duration via agent_runs (ORCH-016 AC-14 DB fallback).
            _build_analyst_ready_comment(repo, work_item_id, branch, task_id=task_id),
            author="analyst",
        )
        notify_approve_requested(task_id)
        result.note = "analysis-in-review"
        logger.info(
            f"Task {task_id}: analyst finished, requested Approved status in Plane"
        )
        return

    questions_path = os.path.join(
        get_worktree_path(repo, branch),
        f"docs/work-items/{work_item_id}/01-questions.md",
    )
    if os.path.isfile(questions_path):
        set_issue_needs_input(work_item_id)
        with open(questions_path, "r") as qf:
            questions_text = qf.read()
        plane_add_comment(
            work_item_id,
            f"\u2753 Analyst \u043d\u0443\u0436\u0434\u0430\u0435\u0442\u0441\u044f \u0432 \u0443\u0442\u043e\u0447\u043d\u0435\u043d\u0438\u0438:\n\n{questions_text}",
            author="analyst",
        )
        send_telegram(
            f"\u2753 {link_for(work_item_id)}: Analyst \u0437\u0430\u0434\u0430\u0451\u0442 \u0432\u043e\u043f\u0440\u043e\u0441\u044b. \u041e\u0442\u0432\u0435\u0442\u044c \u0432 Plane."
        )
        result.note = "analysis-needs-input"
        return

    # No artifacts and no questions.
    plane_add_comment(
        work_item_id,
        "\u26a0\ufe0f Analyst \u0437\u0430\u0432\u0435\u0440\u0448\u0438\u043b\u0441\u044f \u0431\u0435\u0437 \u0430\u0440\u0442\u0435\u0444\u0430\u043a\u0442\u043e\u0432 \u0438 \u0431\u0435\u0437 \u0432\u043e\u043f\u0440\u043e\u0441\u043e\u0432. \u041f\u0440\u043e\u0432\u0435\u0440\u044c\u0442\u0435 \u043b\u043e\u0433.",
        author="analyst",
    )
    result.note = "analysis-empty"


def _handle_qg_failure_rollbacks(
    task_id, current_stage, repo, work_item_id, branch,
    agent, qg_name, reason, result: AdvanceResult,
):
    """All rollback/retry branches from the original launcher, preserved verbatim.

    Only fire on the launcher path (finished_agent is set). The webhook path
    passes finished_agent=None, so none of these agent-specific branches trigger
    — that matches the old plane behavior (it just reported the QG failure).
    """
    # Reviewer REQUEST_CHANGES -> rollback to development + retry (max 3).
    if agent == "reviewer" and "REQUEST_CHANGES" in (reason or ""):
        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"
        retry_count = _developer_retry_count(task_id)
        if retry_count < MAX_DEVELOPER_RETRIES:
            # ORCH-046: embed the verbatim P0/P1 findings into task_desc so the
            # developer agent sees the must-fix points directly (not just a link).
            # extract_review_findings never raises; "" -> graceful link-only fallback.
            review_ref = f"docs/work-items/{work_item_id}/12-review.md"
            review_path = os.path.join(get_worktree_path(repo, branch), review_ref)
            findings = extract_review_findings(review_path)
            head = (
                f"Work item: {work_item_id}\nRepo: {repo}\nBranch: {branch}\n"
                f"Stage: development\nNote: REQUEST_CHANGES from reviewer "
                f"(attempt {retry_count+1}/3)."
            )
            if findings:
                task_desc = (
                    f"{head}\nFindings (P0/P1):\n{findings}\n"
                    f"Полный контекст: {review_ref}"
                )
            else:
                task_desc = f"{head} Fix findings in {review_ref}"
            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}: reviewer REQUEST_CHANGES, enqueued developer "
                f"(job_id={new_job})"
            )
        else:
            send_telegram(
                f"\u26a0\ufe0f {link_for(work_item_id)}: Max developer retries (3) reached. "
                f"Manual intervention needed."
            )
            result.alerted = True
            logger.error(f"Task {task_id}: max retries reached")

    # Tester check_tests_passed FAIL -> rollback to development + retry (max 3).
    if agent == "tester" and qg_name == "check_tests_passed":
        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)
        plane_add_comment(
            work_item_id,
            f"\u274c \u0422\u0435\u0441\u0442\u044b \u043d\u0435 \u043f\u0440\u043e\u0448\u043b\u0438: {reason}. "
            f"Developer \u043f\u0435\u0440\u0435\u0437\u0430\u043f\u0443\u0449\u0435\u043d \u0434\u043b\u044f \u0444\u0438\u043a\u0441\u0430.",
            author="tester",
        )
        retry_count = _developer_retry_count(task_id)
        if retry_count < MAX_DEVELOPER_RETRIES:
            # ORCH-046: embed the gate `reason` plus a verbatim excerpt of the
            # test-report body (pytest output / FAIL rows / Итог) into task_desc.
            # extract_test_failures never raises; "" -> graceful reason+link fallback.
            report_ref = f"docs/work-items/{work_item_id}/13-test-report.md"
            report_path = os.path.join(get_worktree_path(repo, branch), report_ref)
            failures = extract_test_failures(report_path)
            head = (
                f"Work item: {work_item_id}\nRepo: {repo}\nBranch: {branch}\n"
                f"Stage: development\nNote: Tests FAILED. Причина: {reason}."
            )
            if failures:
                task_desc = (
                    f"{head}\nДетали:\n{failures}\n"
                    f"Полный контекст: {report_ref}"
                )
            else:
                task_desc = f"{head} Fix failures described in {report_ref}"
            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}: tester FAIL, enqueued developer (job_id={new_job})"
            )
        else:
            set_issue_blocked(work_item_id)
            send_telegram(
                f"\U0001f6a8 {link_for(work_item_id)}: Tests still failing after 3 developer "
                f"retries. Manual intervention needed."
            )
            result.alerted = True

    # Architect conflict (10-conflict.md exists) -> rollback to analysis.
    if agent == "architect" and qg_name == "check_architecture_done":
        conflict_path = os.path.join(
            get_worktree_path(repo, branch),
            f"docs/work-items/{work_item_id}/10-conflict.md",
        )
        if os.path.isfile(conflict_path):
            update_task_stage(task_id, "analysis")
            notify_stage_change(task_id, current_stage, "analysis")
            plane_notify_stage(work_item_id, current_stage, "analysis")
            result.rolled_back_to = "analysis"
            # ORCH-066 (AC-3): rolled back to analysis -> indicate `Analysis`
            # (degrades to In Progress where the status is not created).
            set_issue_analysis(work_item_id)
            with open(conflict_path, "r") as cf:
                conflict_text = cf.read()[:500]
            plane_add_comment(
                work_item_id,
                f"\u26a0\ufe0f Architect \u043d\u0430\u0448\u0451\u043b \u043a\u043e\u043d\u0444\u043b\u0438\u043a\u0442 \u0441 \u0422\u0417. "
                f"\u0412\u043e\u0437\u0432\u0440\u0430\u0442 \u0432 Analysis.\n\n{conflict_text}",
                author="architect",
            )
            task_desc = (
                f"Work item: {work_item_id}\nRepo: {repo}\nBranch: {branch}\n"
                f"Stage: analysis\nNote: Architect conflict. Revise TRZ. "
                f"See docs/work-items/{work_item_id}/10-conflict.md"
            )
            new_job = enqueue_job("analyst", repo, task_desc, task_id=task_id)
            result.enqueued_agent = "analyst"
            result.enqueued_job_id = new_job
            logger.info(
                f"Task {task_id}: architect conflict, enqueued analyst "
                f"(job_id={new_job})"
            )

    # ORCH-35: deployer staging verdict FAILED -> roll deploy-staging back to development.
    # Staging-провал = код плох; откат на development по образцу БАГ-8 (deploy->development).
    # НЕ трогает ветку check_deploy_status ниже.
    if agent == "deployer" and qg_name == "check_staging_status":
        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_blocked(work_item_id)
        notify_qg_failure(task_id, "deploy-staging", "check_staging_status", reason)
        plane_add_comment(
            work_item_id,
            f"\u274c Staging gate FAILED ({reason}). Rolled back to development. "
            f"Developer \u043d\u0443\u0436\u0435\u043d \u0434\u043b\u044f \u0444\u0438\u043a\u0441\u0430.",
            author="deployer",
        )
        send_telegram(
            f"\U0001f6a8 {link_for(work_item_id)}: Staging FAILED ({reason}). "
            f"Rolled back to development. Needs fix."
        )
        result.alerted = True
        logger.error(
            f"Task {task_id}: deployer staging verdict FAILED, rolled back deploy-staging -> "
            f"development ({reason})"
        )

    # БАГ 8: deployer verdict FAILED -> roll deploy back to development.
    # The launcher's exit_code-based guard (launcher.py:475) never fires because
    # the LLM process exit code is always 0; this gate fires on the machine-readable
    # deploy_status verdict in 14-deploy-log.md instead. Mirrors the launcher block
    # (rollback + set_issue_blocked + notify) but is driven by the VERDICT.
    if agent == "deployer" and qg_name == "check_deploy_status":
        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"
        # ORCH-036: clear the deploy-state sentinels (approve-requested / initiated /
        # result) so the NEXT prod-deploy pass (after the developer fixes and the task
        # returns to `deploy`) is not wedged by Phase B's idempotency-guard reading a
        # STALE `initiated`, nor the finalizer mis-reading a STALE `result`. Markers are
        # keyed by work_item_id (stable across the rollback), so without this they
        # survive into the retry and break re-deploy-after-rollback (AC-4/AC-10).
        try:
            self_deploy.clear_state(repo, work_item_id)
        except Exception as e:  # noqa: BLE001 - defensive (clear_state never-raises anyway)
            logger.warning(f"Task {task_id}: deploy-state clear on deploy-fail failed: {e}")
        # 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(
            work_item_id,
            f"\u274c Deploy FAILED ({reason}). Rolled back to development. "
            f"Developer \u043d\u0443\u0436\u0435\u043d \u0434\u043b\u044f \u0444\u0438\u043a\u0441\u0430.",
            author="deployer",
        )
        send_telegram(
            f"\U0001f6a8 {link_for(work_item_id)}: Deploy FAILED ({reason}). "
            f"Rolled back to development. Needs fix."
        )
        result.alerted = True
        logger.error(
            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 {link_for(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 {link_for(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})"
    )


# ---------------------------------------------------------------------------
# ORCH-022: security sub-gate (secret-scan + dependency audit) on the
# deploy-staging -> deploy edge
# ---------------------------------------------------------------------------
def _handle_security_gate(
    task_id, current_stage, repo, work_item_id, branch, agent, result: AdvanceResult
) -> bool:
    """Run check_security_gate on the deploy-staging -> deploy edge (ORCH-022).

    Runs FIRST among the edge sub-gates — BEFORE the merge-gate and the
    image-freshness rebuild — because it is a cheap read-only scan and we want to
    fail BEFORE the expensive rebase/rebuild (07-infra I-6). Deterministic (no LLM):
    gitleaks (offline secret-scan, fail-closed) + pip-audit (CVE audit, fail-open
    degrade). The machine verdict lives in 17-security-report.md frontmatter.

    Returns True if the gate INTERVENED (the caller must return without advancing):
      * FAIL (secret found / blocking CVE / fail-closed) -> ROLLBACK to development
        (+ developer retry, capped by MAX_DEVELOPER_RETRIES). No merge-lease release
        here: the security-gate runs BEFORE the merge-gate, so the lease is not held
        yet (distinct from the image-freshness rollback). The verbatim findings are
        embedded into the developer's task_desc (ORCH-046 pattern, TC-17).
    Returns False when the gate PASSED (clean, or N/A for a non-self repo with an
    empty scope) so advance_stage proceeds to the merge-gate.
    """
    passed, reason = _run_qg("check_security_gate", repo, work_item_id, branch)
    if passed:
        logger.info(f"Task {task_id}: security-gate passed ({reason})")
        return False

    result.qg_name = "check_security_gate"
    result.qg_passed = False
    result.qg_reason = reason

    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)
    notify_qg_failure(task_id, current_stage, "check_security_gate", reason)
    plane_add_comment(
        work_item_id,
        f"❌ Security-гейт провален ({reason}). Откат на development. "
        f"Developer нужен для фикса (секреты/уязвимые зависимости).",
        author="deployer",
    )
    retry_count = _developer_retry_count(task_id)
    if retry_count < MAX_DEVELOPER_RETRIES:
        # ORCH-046: embed the verbatim findings into task_desc so the developer
        # agent sees the must-fix substance directly (not just a link).
        # extract_security_findings never raises; "" -> graceful link-only fallback.
        report_ref = f"docs/work-items/{work_item_id}/17-security-report.md"
        report_path = os.path.join(get_worktree_path(repo, branch), report_ref)
        findings = extract_security_findings(report_path)
        head = (
            f"Work item: {work_item_id}\nRepo: {repo}\nBranch: {branch}\n"
            f"Stage: development\nNote: Security-гейт провален "
            f"(attempt {retry_count + 1}/{MAX_DEVELOPER_RETRIES}). "
            f"Причина: {reason}."
        )
        if findings:
            task_desc = (
                f"{head}\nFindings:\n{findings}\n"
                f"Полный контекст: {report_ref}"
            )
        else:
            task_desc = f"{head} Fix findings in {report_ref}"
        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}: security-gate FAILED, enqueued developer (job_id={new_job})"
        )
    else:
        set_issue_blocked(work_item_id)
        send_telegram(
            f"\U0001f6a8 {link_for(work_item_id)}: Security-гейт still failing after "
            f"{MAX_DEVELOPER_RETRIES} developer retries ({reason}). "
            f"Manual intervention needed."
        )
        result.alerted = True
    logger.error(
        f"Task {task_id}: security-gate FAILED, rolled back deploy-staging -> "
        f"development ({reason})"
    )
    return True


# ---------------------------------------------------------------------------
# ORCH-058: staging-image freshness sub-gate on the deploy-staging -> deploy edge
# ---------------------------------------------------------------------------
def _handle_image_freshness(
    task_id, current_stage, repo, work_item_id, branch, agent, result: AdvanceResult
) -> bool:
    """Run check_staging_image_fresh on the deploy-staging -> deploy edge (ORCH-058).

    Runs AFTER the merge-gate (validated HEAD finalised) and BEFORE Phase A. The
    sub-check rebuilds the staging image from the validated commit + recreates 8501;
    a green result means the artefact we validate is the exact one that will be
    BUILD-ONCE retagged to prod (AC-4).

    Returns True if the gate INTERVENED (the caller must return without advancing):
      * FAIL (stale / rebuild error / fail-closed) -> ROLLBACK to development
        (+ developer retry, capped by MAX_DEVELOPER_RETRIES) and RELEASE the merge
        lease (the merge-gate held it on its PASS). Mirrors the merge-gate rollback.
    Returns False when the gate PASSED (fresh, or N/A for a non-self repo) so
    advance_stage proceeds to Phase A. On a PASS the merge lease stays HELD until
    the actual merge (released on PR-merged webhook / deploy->done / rollback).
    """
    passed, reason = _run_qg("check_staging_image_fresh", repo, work_item_id, branch)
    if passed:
        logger.info(f"Task {task_id}: image-freshness passed ({reason})")
        return False

    result.qg_name = "check_staging_image_fresh"
    result.qg_passed = False
    result.qg_reason = reason

    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)
    # The merge-gate held the lease on its PASS; freshness failed before the merge,
    # so release it (holder-aware no-op if a different task already owns it).
    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 image-freshness fail failed: {e}")
    notify_qg_failure(task_id, current_stage, "check_staging_image_fresh", reason)
    plane_add_comment(
        work_item_id,
        f"❌ Staging-образ не свеж ({reason}). Откат на 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: Staging image freshness 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}: image-freshness FAILED, enqueued developer (job_id={new_job})"
        )
    else:
        set_issue_blocked(work_item_id)
        send_telegram(
            f"\U0001f6a8 {link_for(work_item_id)}: Staging image freshness still failing after "
            f"{MAX_DEVELOPER_RETRIES} developer retries ({reason}). "
            f"Manual intervention needed."
        )
        result.alerted = True
    logger.error(
        f"Task {task_id}: image-freshness FAILED, rolled back deploy-staging -> "
        f"development ({reason})"
    )
    return True


# ---------------------------------------------------------------------------
# ORCH-036: executable self-deploy (Phase A/B/C)
# ---------------------------------------------------------------------------
def _handle_self_deploy_phase_a(
    task_id, current_stage, repo, work_item_id, branch, result: AdvanceResult
):
    """Phase A — advance to `deploy` and request a manual approve (no prod deploy).

    Staging is green and the branch is mergeable; for the self-hosting repo we do
    NOT auto-deploy to prod. Move the task onto the `deploy` stage (so a later
    human "Confirm Deploy" lands there -> Phase B), set the issue approval-pending
    and ask the human to flip the status to "Confirm Deploy" (ORCH-059: the
    dedicated prod-deploy trigger, distinct from the human-gate "Approved"). A
    restart-safe `approve-requested` marker records that Phase A ran. The merge
    lease stays HELD.
    """
    update_task_stage(task_id, "deploy")
    notify_stage_change(task_id, current_stage, "deploy")
    result.advanced = True
    result.to_stage = "deploy"
    result.note = "self-deploy-approval-pending"

    if work_item_id:
        # ORCH-066 (AC-6/AC-13): Phase A approval-pending is now `Awaiting Deploy`,
        # which discharges `In Review` of the deploy-approval meaning (In Review
        # stays for analyst BRD/review approve-pending only). Degrades to In Review
        # where the status is not created.
        set_issue_awaiting_deploy(work_item_id)
    # ORCH-036: belt-and-suspenders — wipe any STALE deploy-state markers before
    # arming a fresh approve. A prior FAILED pass clears on rollback, but clearing
    # here too guarantees the entry to every new prod-deploy pass starts clean
    # (e.g. after a crash/manual intervention), so `initiated`/`result` from an
    # earlier attempt can never leak into this one.
    self_deploy.clear_state(repo, work_item_id)
    self_deploy.write_marker(
        repo, work_item_id, self_deploy.APPROVE_REQUESTED, content=str(time.time())
    )
    if work_item_id:
        plane_add_comment(
            work_item_id,
            "\U0001f7e1 Staging зелёный. Требуется ручное подтверждение ПРОД-деплоя: "
            "смените статус задачи на «Confirm Deploy», чтобы запустить деплой в прод "
            "(8500). Статус «Approved» прод-деплой НЕ запускает.",
            author="deployer",
        )
    send_telegram(
        f"\U0001f7e1 {link_for(work_item_id)}: staging OK. Ждёт подтверждения ПРОД-деплоя "
        f"(смените статус на «Confirm Deploy»)."
    )
    logger.info(
        f"Task {task_id}: self-deploy Phase A — advanced to deploy, "
        f"approval-pending (awaiting human Approved)"
    )


def _handle_self_deploy_phase_b(task_id, repo, work_item_id, branch, result: AdvanceResult):
    """Phase B — a human Approved initiates the DETACHED prod deploy (idempotent).

    Idempotency-guard: if the `initiated` marker already exists (double Approved /
    duplicate webhook, R-4) this is a no-op. Otherwise launch the detached host
    deploy, and ONLY on success record `initiated` + enqueue the finalizer (so a
    failed launch can be retried by re-approving). Returns without advancing — the
    finalizer (Phase C) records the verdict once the hook finishes.
    """
    if self_deploy.has_marker(repo, work_item_id, self_deploy.INITIATED):
        result.note = "self-deploy-already-initiated"
        logger.info(
            f"Task {task_id}: prod deploy already initiated; ignoring repeat Approved"
        )
        return

    ok, msg = self_deploy.initiate_deploy(repo, work_item_id, branch)
    if not ok:
        result.note = f"self-deploy-initiate-failed: {msg}"
        if work_item_id:
            plane_add_comment(
                work_item_id,
                f"⚠️ Не удалось запустить прод-деплой: {msg}. "
                "Повторите approve после устранения причины.",
                author="deployer",
            )
        send_telegram(f"⚠️ {link_for(work_item_id)}: прод-деплой не запустился: {msg}")
        logger.error(f"Task {task_id}: self-deploy initiate failed: {msg}")
        return

    self_deploy.write_marker(
        repo, work_item_id, self_deploy.INITIATED, content=str(time.time())
    )
    # ORCH-066 (AC-7): the prod deploy is now in flight -> indicate `Deploying`
    # (degrades to In Progress where the status is not created).
    if work_item_id:
        set_issue_deploying(work_item_id)
    task_desc = (
        f"Work item: {work_item_id}\nRepo: {repo}\nBranch: {branch}\n"
        f"Stage: deploy\nNote: deploy-finalize poll (prod self-deploy initiated)."
    )
    new_job = enqueue_job(
        "deploy-finalizer", repo, task_desc, task_id=task_id,
        available_at_delay_s=settings.deploy_finalize_delay_s,
    )
    result.enqueued_agent = "deploy-finalizer"
    result.enqueued_job_id = new_job
    result.note = "self-deploy-initiated"
    if work_item_id:
        plane_add_comment(
            work_item_id,
            "\U0001f680 Прод-деплой стартовал (detached host-процесс). "
            "Вердикт будет зафиксирован после health-check.",
            author="deployer",
        )
    send_telegram(f"\U0001f680 {link_for(work_item_id)}: прод-деплой стартовал. Жду результат.")
    logger.info(
        f"Task {task_id}: self-deploy Phase B — detached deploy initiated, "
        f"finalizer enqueued (job_id={new_job})"
    )


def _deploy_finalize_defer_count(task_id: int) -> int:
    """How many times this task's finalizer has already deferred (restart-safe).

    Counted from the persisted jobs queue by the defer marker in task_content
    (mirrors _merge_defer_count), so a service restart never resets the budget.
    """
    conn = get_db()
    n = conn.execute(
        "SELECT COUNT(*) FROM jobs WHERE task_id=? AND task_content LIKE '%deploy-finalize defer%'",
        (task_id,),
    ).fetchone()[0]
    conn.close()
    return n


def _hold_main_regressed(
    task_id, repo, work_item_id, branch, guard_msg: str, result: AdvanceResult
) -> bool:
    """HOLD the task because the regression guard found neighbouring code missing.

    ORCH-073 Р-4 (FR-5 / AC-5): the deployed SHA IS in `main` (FR-1 passed) but a
    declarative marker of a recently-merged task is gone -> a phantom-merge / rebase
    rolled back sibling code. Reaction is ALERT-only + HOLD (Telegram + Plane
    ``set_issue_blocked`` + comment), task stays on `deploy` (NOT done), NO rollback
    to development (an infra defect, not a code fault — symmetric to the not-merged
    HOLD). Returns ``True`` (INTERVENED). Never breaks the HOLD on a notify error.
    """
    merge_gate.note_main_regressed_alert(work_item_id)
    msg = (
        f"main regressed: {guard_msg} (repo={repo}, branch={branch}, "
        f"wi={work_item_id}). Соседний код пропал из `main` — задача удержана на "
        f"`deploy` (НЕ done). Нужно ручное восстановление кода."
    )
    logger.warning(f"Task {task_id}: {msg}")
    if work_item_id:
        try:
            set_issue_blocked(work_item_id)
        except Exception as e:  # noqa: BLE001 - never break the HOLD
            logger.warning(f"Task {task_id}: set_issue_blocked failed: {e}")
        try:
            plane_add_comment(
                work_item_id,
                "\U0001f6a8 Регресс `main`: " + guard_msg + ". Код соседней задачи "
                "пропал из `main`. Задача удержана на `deploy` (НЕ done) — нужно "
                "восстановить код и повторить approve.",
                author="deployer",
            )
        except Exception as e:  # noqa: BLE001 - never break the HOLD
            logger.warning(f"Task {task_id}: plane regressed comment failed: {e}")
    try:
        send_telegram(f"\U0001f6a8 {msg}")
    except Exception as e:  # noqa: BLE001 - never break the HOLD
        logger.warning(f"Task {task_id}: main-regressed telegram failed: {e}")
    result.alerted = True
    result.note = "main-regressed-hold"
    result.advanced = False
    return True


def _hold_pr_create_failed(
    task_id, repo, work_item_id, branch, reason: str, result: AdvanceResult
) -> bool:
    """HOLD the task because the open code-PR could not be ensured (ORCH-082 Р-3).

    FR-2/FR-4 (AC-5/AC-7): ``ensure_open_pr`` returned ``"failed"`` (Gitea unreachable /
    HTTP error) — there is no open code-PR and one could not be created. Symmetric to the
    not-merged / regressed HOLD: task stays on ``deploy`` (NOT done), NO rollback to
    development, ALERT-only (Telegram + Plane ``set_issue_blocked`` + comment). The HOLD
    text MUST be distinguishable from the not-merged HOLD so the operator sees the cause is
    "could not CREATE the PR" (infra), not "could not MERGE an existing one". Returns
    ``True`` (INTERVENED). Never breaks the HOLD on a notify error; ``failed`` is a
    structured outcome, not a propagated exception (INV-1).
    """
    merge_gate.note_not_merged_alert(work_item_id)  # reuse the counter-notifier.
    msg = (
        f"PR создать не удалось: {reason} (repo={repo}, branch={branch}, "
        f"wi={work_item_id}). Открытый код-PR отсутствует и не создан — задача "
        f"удержана на `deploy` (НЕ done). Нужно проверить доступность Gitea / создать PR."
    )
    logger.warning(f"Task {task_id}: {msg}")
    if work_item_id:
        try:
            set_issue_blocked(work_item_id)
        except Exception as e:  # noqa: BLE001 - never break the HOLD
            logger.warning(f"Task {task_id}: set_issue_blocked failed: {e}")
        try:
            plane_add_comment(
                work_item_id,
                "\U0001f6a8 PR создать не удалось: " + reason + ". Открытый код-PR "
                "отсутствует — задача удержана на `deploy` (НЕ done). Проверьте "
                "доступность Gitea / создайте PR вручную и повторите approve.",
                author="deployer",
            )
        except Exception as e:  # noqa: BLE001 - never break the HOLD
            logger.warning(f"Task {task_id}: plane pr-create-failed comment failed: {e}")
    try:
        send_telegram(f"\U0001f6a8 {msg}")
    except Exception as e:  # noqa: BLE001 - never break the HOLD
        logger.warning(f"Task {task_id}: pr-create-failed telegram failed: {e}")
    result.alerted = True
    result.note = "pr-create-failed-hold"
    result.advanced = False
    return True


def _handle_merge_verify(task_id, repo, work_item_id, branch, result: AdvanceResult) -> bool:
    """ORCH-071 merge-verify under-gate on the `deploy -> done` edge.

    Returns:
      * ``True``  -> INTERVENED (HOLD): the merge is NOT confirmed -> alert +
        ``set_issue_blocked`` (Plane non-terminal), task stays on `deploy`, NO
        ``done``, NO rollback to development (not-merged is an INFRA defect, not a
        code fault -> ALERT-only, FR-3). The caller returns without advancing. A
        later re-drive (reaper / reconciler / re-approve) re-evaluates and, once the
        merge is fixed, lets the task advance to `done`.
      * ``False`` -> the merge is CONFIRMED (or the under-gate does not apply for
        this repo / kill-switch off) -> ``advance_stage`` proceeds to `done`
        unchanged (happy-path AC-4 / AC-4b).

    Steps (D5):
      1. Conditionality (FR-5): not applicable -> return False (1:1 prior behaviour).
      2. Resolve the validated SHA; run the deterministic merge-actor
         ``merge_gate.merge_pr`` (no-op if already merged, INV-5).
      3. ``merge_gate.verify_merged_to_main`` -> confirmed?
         * yes -> stamp ``merged_to_main: true``, return False (advance).
         * no  -> alert + Blocked + stamp ``merged_to_main: false``, return True (HOLD).

    Wrapped never-raise (INV-1/AC-7): any internal error is treated as "not
    confirmed" (HOLD + alert), never a propagated exception into ``advance_stage``.
    """
    try:
        if not merge_gate.merge_verify_applies(repo):
            return False  # non-self / kill-switch off -> behave exactly as before.

        from . import image_freshness
        sha = image_freshness.validated_revision(repo, branch)

        # ORCH-082 (Р-2 / FR-2): guarantee an open code-PR (head==branch, base==main)
        # BEFORE the deterministic merge_pr. The pipeline never guaranteed the branch
        # had one at merge time (PRs are created only on the developer path with a fresh
        # commit) -> a PR-less branch hit merge_pr "no open PR" -> a FALSE HOLD (ORCH-074).
        # `created`/`existed` -> proceed unchanged; `failed` -> honest HOLD with a
        # distinguishable text (NOT the not-merged HOLD). ORCH-073's SHA-in-main proof
        # below is untouched and stays authoritative. Kill-switch off -> 1:1 prior path.
        if settings.merge_verify_autocreate_pr_enabled:
            pr_status, pr_detail = merge_gate.ensure_open_pr(repo, branch)
            logger.info(
                f"Task {task_id}: merge-verify ensure_open_pr -> {pr_status} ({pr_detail})"
            )
            if pr_status == "failed":
                return _hold_pr_create_failed(
                    task_id, repo, work_item_id, branch, pr_detail, result
                )
            # "created" | "existed" -> proceed normally to merge_pr.

        # Deterministic merge-actor (no-op if the PR is already merged, INV-5/AC-9).
        merged_ok, merge_msg = merge_gate.merge_pr(repo, branch)
        logger.info(
            f"Task {task_id}: merge-verify merge_pr -> ok={merged_ok} ({merge_msg})"
        )

        confirmed = merge_gate.verify_merged_to_main(repo, branch, sha)
        if confirmed:
            # ORCH-073 Р-4 (FR-5): secondary main-integrity regression guard. The
            # deployed SHA is in main (FR-1), but a CHANGELOG-rebase / phantom-merge
            # could still have rolled back a NEIGHBOURING task's code. Verify the
            # declarative marker set is intact; a deterministic miss -> HOLD + alert
            # (NOT done, no rollback — an infra defect, not a code fault). Fail-OPEN
            # on a git error of the guard itself (SHA-in-main remains the primary
            # gate). Honours the same scope/kill-switch as the under-gate.
            if settings.regression_guard_enabled:
                guard_ok, guard_msg = merge_gate.check_main_regression(repo, branch)
                if not guard_ok:
                    return _hold_main_regressed(
                        task_id, repo, work_item_id, branch, guard_msg, result
                    )

            merge_gate.note_merge_verified()
            try:
                self_deploy.record_merged_to_main(repo, work_item_id, branch, True)
            except Exception as e:  # noqa: BLE001 - observability best-effort
                logger.warning(f"Task {task_id}: record merged_to_main(true) failed: {e}")
            logger.info(f"Task {task_id}: merge-verify CONFIRMED -> deploy->done allowed")
            return False

        # Not confirmed -> alert + HOLD (no done, no rollback).
        merge_gate.note_not_merged_alert(work_item_id)
        try:
            self_deploy.record_merged_to_main(repo, work_item_id, branch, False)
        except Exception as e:  # noqa: BLE001 - observability best-effort
            logger.warning(f"Task {task_id}: record merged_to_main(false) failed: {e}")
        msg = (
            f"deploy succeeded but not merged: {work_item_id} (repo={repo}, "
            f"branch={branch}). `main` НЕ получил commit задачи — задача удержана "
            f"на `deploy` (НЕ done). Нужно ручное вмешательство."
        )
        logger.warning(f"Task {task_id}: {msg}")
        if work_item_id:
            try:
                set_issue_blocked(work_item_id)
            except Exception as e:  # noqa: BLE001 - never break the HOLD
                logger.warning(f"Task {task_id}: set_issue_blocked failed: {e}")
            try:
                plane_add_comment(
                    work_item_id,
                    "\U0001f6a8 Deploy прошёл, но PR НЕ влит в `main` "
                    f"(merge: {merge_msg}). Задача удержана на `deploy` (НЕ done). "
                    "Нужно влить PR вручную и повторить approve.",
                    author="deployer",
                )
            except Exception as e:  # noqa: BLE001 - never break the HOLD
                logger.warning(f"Task {task_id}: plane not-merged comment failed: {e}")
        try:
            send_telegram(f"\U0001f6a8 {msg}")
        except Exception as e:  # noqa: BLE001 - never break the HOLD
            logger.warning(f"Task {task_id}: not-merged telegram failed: {e}")
        result.alerted = True
        result.note = "merge-not-verified-hold"
        result.advanced = False
        return True
    except Exception as e:  # noqa: BLE001 - never-raise contract (INV-1/AC-7)
        # Any internal error -> treat as "not confirmed" -> HOLD + alert, never crash.
        logger.error(f"Task {task_id}: _handle_merge_verify error: {e}")
        try:
            merge_gate.note_not_merged_alert(work_item_id)
            send_telegram(
                f"\U0001f6a8 {link_for(work_item_id)}: ошибка merge-verify ({e}). "
                f"Задача удержана на `deploy` (НЕ done)."
            )
        except Exception:  # noqa: BLE001 - best-effort alert
            pass
        result.alerted = True
        result.note = f"merge-verify-error: {e}"
        result.advanced = False
        return True


def run_deploy_finalizer(job: dict):
    """Phase C — deterministic finalizer (reserved-agent `deploy-finalizer`, no LLM).

    Claimed by the worker in the NEW container after the prod restart. Reads the
    `result` sentinel (hook exit-code written by the host wrapper):
      * not written yet & budget left -> DEFER (re-queue with a delay);
      * budget exhausted              -> set_issue_blocked + Telegram (anti-livelock);
      * present                       -> map exit-code -> deploy_status, write
        14-deploy-log.md, then advance_stage(finished_agent="deployer") so the
        EXISTING contracts fire: SUCCESS -> terminal-sync deploy->done + release
        lease; FAILED -> БАГ-8 rollback deploy->development + set_issue_blocked.
    Never raises into the caller (the launcher marks the job done/failed).
    """
    task_id = job.get("task_id")
    repo = job.get("repo")
    conn = get_db()
    row = conn.execute(
        "SELECT work_item_id, branch FROM tasks WHERE id=?", (task_id,)
    ).fetchone()
    conn.close()
    if not row:
        logger.error(f"deploy-finalizer: no task row for task_id={task_id}")
        return
    work_item_id, branch = row[0], row[1]

    present, code = self_deploy.read_result(repo, work_item_id)
    if not present:
        defers = _deploy_finalize_defer_count(task_id)
        if defers < settings.deploy_finalize_max_attempts:
            task_desc = (
                f"Work item: {work_item_id}\nRepo: {repo}\nBranch: {branch}\n"
                f"Stage: deploy\nNote: deploy-finalize defer "
                f"(attempt {defers + 1}/{settings.deploy_finalize_max_attempts}) — "
                f"deploy result not ready, retrying after {settings.deploy_finalize_delay_s}s."
            )
            new_job = enqueue_job(
                "deploy-finalizer", repo, task_desc, task_id=task_id,
                available_at_delay_s=settings.deploy_finalize_delay_s,
            )
            logger.info(
                f"Task {task_id}: deploy result not ready, finalizer deferred "
                f"(job_id={new_job}, attempt {defers + 1}/{settings.deploy_finalize_max_attempts})"
            )
        else:
            if work_item_id:
                set_issue_blocked(work_item_id)
            send_telegram(
                f"\U0001f6a8 {link_for(work_item_id)}: deploy result не появился после "
                f"{settings.deploy_finalize_max_attempts} попыток. Нужно ручное вмешательство."
            )
            logger.error(
                f"Task {task_id}: deploy-finalize defer attempts exhausted "
                f"({settings.deploy_finalize_max_attempts})"
            )
        return

    # Result present -> deterministic verdict.
    status = self_deploy.map_exit_code_to_status(code)
    self_deploy.write_deploy_log(repo, work_item_id, branch, code, status)
    logger.info(
        f"Task {task_id}: deploy finalized, hook exit={code} -> deploy_status={status}"
    )
    if status == "SUCCESS" and work_item_id:
        plane_add_comment(
            work_item_id,
            f"✅ Прод-деплой успешен (health-check OK, exit {code}).",
            author="deployer",
        )
        send_telegram(f"✅ {link_for(work_item_id)}: прод-деплой успешен (exit {code}).")

    # Drive the EXISTING deploy contracts via the gate verdict we just wrote.
    advance_stage(
        task_id=task_id,
        current_stage="deploy",
        repo=repo,
        work_item_id=work_item_id,
        branch=branch,
        finished_agent="deployer",
    )


def run_post_deploy_monitor(job: dict):
    """ORCH-021 — one post-deploy monitor tick (reserved-agent, no LLM).

    A deterministic tick modelled on ``run_deploy_finalizer``: it does ONE probe
    of the prod instance, appends to the persisted ``series`` (restart-safe
    streak/5xx counters), classifies, and then either RE-QUEUES itself with a
    delay (window not over and still HEALTHY) or FINISHES the window (DEGRADED ->
    reaction; window expired -> HEALTHY). Observation happens entirely AFTER the
    terminal ``done`` — it never touches ``STAGE_TRANSITIONS`` / ``QG_CHECKS`` and
    never restarts the prod orchestrator container itself (AC-8 / AC-12).

    never-raise into the caller (the launcher marks the job done/failed); each
    branch is individually defensive.
    """
    task_id = job.get("task_id")
    repo = job.get("repo")
    try:
        conn = get_db()
        row = conn.execute(
            "SELECT work_item_id, branch FROM tasks WHERE id=?", (task_id,)
        ).fetchone()
        conn.close()
    except Exception as e:  # noqa: BLE001 - never-raise
        logger.error(f"post-deploy-monitor: db error for task_id={task_id}: {e}")
        return
    if not row:
        logger.error(f"post-deploy-monitor: no task row for task_id={task_id}")
        return
    work_item_id, branch = row[0], row[1]

    # AC-15: a finished window is a no-op (defends against a duplicate job).
    if post_deploy.has_marker(repo, work_item_id, post_deploy.DONE):
        logger.info(f"post-deploy-monitor: {work_item_id} already done (no-op)")
        return

    # One probe -> append -> classify (restart-safe via the persisted series).
    probe = post_deploy.probe_signals(settings.post_deploy_base_url)
    series = post_deploy.append_probe(repo, work_item_id, probe)
    verdict = post_deploy.classify(
        series,
        settings.post_deploy_fail_threshold,
        settings.post_deploy_5xx_threshold,
    )
    ticks = len(series)
    budget = post_deploy.max_ticks()
    logger.info(
        f"post-deploy-monitor: {work_item_id} tick {ticks}/{budget} "
        f"probe=[{probe.detail}] verdict={verdict}"
    )

    # HEALTHY and window not exhausted -> defer the next tick (worker stays free).
    if verdict == post_deploy.HEALTHY and ticks < budget:
        task_desc = (
            f"Work item: {work_item_id}\nRepo: {repo}\nBranch: {branch}\n"
            f"Stage: post-deploy\nNote: post-deploy monitor tick {ticks + 1} "
            f"(healthy so far; re-poll after {settings.post_deploy_interval_s}s)."
        )
        enqueue_job(
            "post-deploy-monitor", repo, task_desc, task_id=task_id,
            available_at_delay_s=settings.post_deploy_interval_s,
        )
        return

    checks_total = ticks
    checks_failed = sum(1 for r in series if not r.get("health_ok"))

    # HEALTHY and window exhausted -> clean finish (BR-6 / AC-17).
    if verdict == post_deploy.HEALTHY:
        post_deploy.write_post_deploy_log(
            repo, work_item_id, branch, post_deploy.HEALTHY, post_deploy.NONE,
            settings.post_deploy_window_s, checks_total, checks_failed,
        )
        post_deploy.mark_done(repo, work_item_id)
        # ORCH-066 (AC-10): the post-deploy window closed clean -> terminal Done.
        if work_item_id:
            try:
                set_issue_done(work_item_id)
            except Exception as e:  # noqa: BLE001 - never break the tick
                logger.warning(f"post-deploy: set Done failed for {work_item_id}: {e}")
        _notify_post_deploy(
            work_item_id,
            f"✅ {work_item_id}: пост-деплой окно завершено чисто "
            f"(HEALTHY, {checks_total} опросов).",
        )
        return

    # DEGRADED -> decide + execute the reaction (§5), write artefact, finish.
    action = post_deploy.decide_action(repo, verdict)
    action_taken = post_deploy.ALERT_ONLY
    if action == post_deploy.ROLLBACK:
        # Non-self repo + auto policy: run the --rollback hook synchronously (the
        # target is NOT the orchestrator, so its restart is safe for the pipeline).
        exit_code, detail = post_deploy.run_rollback(repo)
        action_taken = post_deploy.map_rollback_exit_code(exit_code)
        if action_taken == post_deploy.ROLLBACK_OK:
            _notify_post_deploy(
                work_item_id,
                f"⚠️ {work_item_id}: пост-деплой DEGRADED -> авто-rollback выполнен "
                f"(exit {exit_code}).",
            )
        else:
            # AC-9: a failed rollback escalates loudly for manual intervention.
            _notify_post_deploy(
                work_item_id,
                f"🚨 {work_item_id}: пост-деплой DEGRADED -> авто-rollback ПРОВАЛИЛСЯ "
                f"(exit {exit_code}: {detail}). Нужно ручное вмешательство.",
            )
    else:
        # ALERT_ONLY: self-hosting ALWAYS lands here — the tick NEVER auto-rolls-back
        # or restarts the prod orchestrator container (BR-5 / AC-8). Loud alert +
        # manual-approve request (mirrors deploy Phase A CTA).
        action_taken = post_deploy.ALERT_ONLY
        _notify_post_deploy(
            work_item_id,
            f"🚨 {work_item_id}: пост-деплой DEGRADED ({checks_failed}/{checks_total} "
            f"провалов). Требуется ручной approve отката — авто-rollback для "
            f"self-hosting запрещён (BR-5).",
        )

    # ORCH-066 (AC-11/AC-12): a confirmed degradation -> indicate `Blocked` for
    # manual intervention. This is INDICATION ONLY — the tick NEVER restarts /
    # rolls back the prod container (self-hosting stays ALERT_ONLY, BR-5).
    if work_item_id:
        try:
            set_issue_blocked(work_item_id)
        except Exception as e:  # noqa: BLE001 - never break the tick
            logger.warning(f"post-deploy: set Blocked failed for {work_item_id}: {e}")

    post_deploy.write_post_deploy_log(
        repo, work_item_id, branch, post_deploy.DEGRADED, action_taken,
        settings.post_deploy_window_s, checks_total, checks_failed,
    )
    post_deploy.mark_done(repo, work_item_id)


def _notify_post_deploy(work_item_id: str, message: str) -> None:
    """Best-effort Telegram + Plane notification for a post-deploy event (AC-17).

    Never raises — a notification failure must not wedge the monitor tick.
    """
    try:
        send_telegram(message)
    except Exception as e:  # noqa: BLE001 - never break the tick
        logger.warning(f"post-deploy notify telegram failed for {work_item_id}: {e}")
    if work_item_id:
        try:
            plane_add_comment(work_item_id, message, author="deployer")
        except Exception as e:  # noqa: BLE001 - never break the tick
            logger.warning(f"post-deploy notify plane failed for {work_item_id}: {e}")
