"""Deterministic staging-runner (ORCH-115).

The `deploy-staging` stage for the self-hosting ``orchestrator`` repo was driven by
the LLM ``deployer`` agent, but the actual work is purely deterministic
(``.openclaw/agents/deployer.md`` steps 1-4): run the staging suite, map its
**exit-code** to a verdict (``0 -> SUCCESS``, else ``FAILED``), write
``15-staging-log.md`` and commit it. This leaf replaces that LLM consultation with
deterministic code, intercepted in ``launcher.launch_job`` BEFORE ``_spawn`` (the
``deploy-finalizer`` / ``post-deploy-monitor`` reserved-agent precedent,
``launcher.py:389/394``).

What is and is NOT changed (NFR-1, the critical invariant):
  * UNCHANGED — the artifact contract (``15-staging-log.md`` with
    ``staging_status: SUCCESS|FAILED``), the gate ``check_staging_status`` /
    ``_parse_staging_status``, ``STAGE_TRANSITIONS``, the DB schema. This module
    replaces only the *producer* of the artifact, never the gate that reads it.
  * NEW — a deterministic producer + a launch_job intercept. Under a kill-switch +
    repo-scope CSV; fail-safe back to the LLM path when off / out of scope.

This module is a **leaf** (mirror of ``self_deploy`` / ``proc_group`` /
``staging_verdict``): it imports only ``config`` / ``logging`` / ``proc_group`` at
module load; ``db`` / ``git_worktree`` / ``self_deploy.map_exit_code_to_status`` /
``qg.checks`` / ``stage_engine.advance_stage`` / ``notifications`` are imported
LAZILY inside functions so the heavy ``stage_engine`` is never pulled at import and
no import cycle forms (pattern: ``transition_lease`` / ``merge_gate``). Every public
function honours a **never-raise** contract (AC-7): a staging-infra hiccup can never
crash the worker / wedge the queue.

Two-level outcome (D5 — the key safety decision, anti ORCH-110):
  * the suite EXECUTED (a real exit-code, 0 or non-zero) -> trust the code:
    ``0 -> SUCCESS -> advance``; ``!=0 -> FAILED -> the existing rollback
    deploy-staging -> development`` (same developer-retry path as a FAILED LLM
    verdict). ORCH-061 infra-tolerance is already INSIDE ``staging_check.py`` — the
    runner never re-judges it (BR-4).
  * the suite did NOT execute (tool-error: spawn-error / timeout / ``returncode is
    None``) -> an infra fault, NOT a code fault -> a bounded DEFER (re-queue a fresh
    ``deployer`` job with a delay + a restart-safe marker). On budget exhaustion ->
    fail-closed ``FAILED`` + advance + alert. So the runner NEVER does a silent
    advance / false green, and NEVER wedges the queue, but does NOT burn a
    developer-retry on transient infra.
"""

import logging
import shlex
import subprocess
import time

from .config import settings
from . import proc_group

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

# Platform service literal — the staging compose service the suite runs inside.
# Already an accepted platform literal (mirror image_freshness._STAGING_SERVICE);
# NOT a host hardcode (test_no_host_hardcodes forbids host IP/home/hostname only).
STAGING_SERVICE = "orchestrator-staging"

# ORCH-123 (D3): deterministic stderr markers that prove a PERMANENT environment
# defect (no docker / container down / daemon unreachable) — NOT a transient infra
# hiccup and NEVER a code-fail. Scanned lower-cased; mirror of merge_gate's
# scope-guard marker approach. Recognised suite exit-codes WITHOUT any of these
# markers are always trusted as `suite-ran` (anti-over-tolerance, BR-3).
_ENV_MARKERS = (
    "command not found",
    "executable file not found",
    "no such container",
    "is not running",
    "cannot connect to the docker daemon",
    "is the docker daemon running",
)

# ORCH-123 (D5): last prod-like preflight verdict of the host-side staging channel
# (observed at startup; surfaced in snapshot()). `ok=None` -> not probed / n/a.
_PREFLIGHT_STATE: dict = {"ok": None, "reason": "not-probed"}
_PREFLIGHT_TIMEOUT_S = 20

# Default wall-clock budget for the docker-exec staging suite (D9). Kept <= the LLM
# staging window it replaces so Σ(work on the deploy-staging edge) does not grow and
# the cross-cutting reaper invariant (ORCH-065/109/110) holds WITHOUT touching
# reaper_max_running_s.
_DEFAULT_TIMEOUT_S = 600

_GIT_TIMEOUT = 60

# Restart-safe DEFER marker (counted from the persisted jobs queue, mirror of
# stage_engine._merge_infra_retry_count). Embedded verbatim in the re-queued job's
# task_content so a service restart never resets the infra-retry budget.
_INFRA_RETRY_MARKER = "staging-runner infra-retry"

# In-process observability counters (mirror merge_gate._MERGE_GATE_COUNTERS, ORCH-110).
# ORCH-123 (D8): `failed` now counts ONLY a real code-fail (suite ran, exit != 0);
# `permanent_env` counts infra-HOLD events (a permanent environment defect, OR a
# transient-infra DEFER budget exhausted) so /queue distinguishes the THREE
# non-success classes: code-fail (`failed`) vs transient-infra (`deferred`) vs
# permanent-env / infra-HOLD (`permanent_env`).
_STAGING_RUNNER_COUNTERS: dict = {
    "runs": 0,          # run_staging_gate entered
    "success": 0,       # suite ran, exit 0 -> SUCCESS
    "failed": 0,        # suite ran non-zero -> FAILED (a real code-fail)
    "tool_error": 0,    # suite did NOT execute (spawn-error / timeout / None)
    "deferred": 0,      # bounded transient-infra DEFER (re-queued)
    "permanent_env": 0, # infra-HOLD: permanent env defect OR transient budget exhausted
}


def _bump(key: str) -> None:
    """Increment an observability counter. Never raises."""
    try:
        _STAGING_RUNNER_COUNTERS[key] += 1
    except Exception:  # noqa: BLE001 - observability must never break a decision
        pass


# ---------------------------------------------------------------------------
# Conditionality (D8 / FR-6 / AC-6)
# ---------------------------------------------------------------------------
def applies(repo: str) -> bool:
    """Whether the deterministic staging-runner is REAL for ``repo``.

    Mirrors ``self_deploy_applies`` / ``coverage_gate``:
      * ``staging_runner_enabled=False`` -> always False (global kill-switch); the
        legacy LLM-deployer path runs on deploy-staging via ``_spawn``.
      * ``staging_runner_repos`` (CSV) non-empty -> only the listed repos.
      * empty CSV -> only the self-hosting repo (``orchestrator``), which is the only
        one with a staging instance.
    Checked FIRST in ``should_intercept`` (local, no network, no DB) so a disabled
    flag costs nothing. never-raise -> False (fail-safe to the prior LLM path).
    """
    try:
        if not settings.staging_runner_enabled:
            return False
        raw = (settings.staging_runner_repos or "").strip()
        if raw:
            allowed = {r.strip().lower() for r in raw.split(",") if r.strip()}
            return (repo or "").strip().lower() in allowed
        # Lazy import keeps this module a leaf (no qg import at module load).
        from .qg.checks import is_self_hosting_repo
        return is_self_hosting_repo(repo)
    except Exception as e:  # noqa: BLE001 - never-raise contract
        logger.warning("staging_runner.applies error for %s: %s", repo, e)
        return False


def should_intercept(job: dict) -> bool:
    """True iff this ``deployer`` job is the deterministic staging-suite job (D1).

    The discriminator of "staging vs prod" is the TASK STAGE, not the role name
    (``deployer`` owns both ``deploy-staging`` and ``deploy``): intercept iff
    ``agent == "deployer"`` AND the task's ``tasks.stage == "deploy-staging"`` AND
    ``applies(repo)``. For self-hosting the prod ``deploy`` edge runs Phase A (no
    deployer spawn); for non-self repos ``applies`` is False, so a non-self prod
    ``deployer`` job is never intercepted (R-1 / TC-06).

    never-raise -> False (a DB-lookup failure falls through to ``_spawn``, fail-safe
    to the prior LLM path).
    """
    try:
        if (job.get("agent") or "") != "deployer":
            return False
        # applies() FIRST (local, no DB): disabled flag -> zero DB overhead.
        if not applies(job.get("repo")):
            return False
        task_id = job.get("task_id")
        if task_id is None:
            return False
        from .db import get_db
        conn = get_db()
        row = conn.execute("SELECT stage FROM tasks WHERE id=?", (task_id,)).fetchone()
        conn.close()
        if not row:
            return False
        return (row[0] or "") == "deploy-staging"
    except Exception as e:  # noqa: BLE001 - never-raise contract
        logger.warning("staging_runner.should_intercept error: %s", e)
        return False


# ---------------------------------------------------------------------------
# Suite execution (D3 / FR-2 / NFR-3 / AC-8 / AC-9)
# ---------------------------------------------------------------------------
def _ssh_target() -> str:
    """ssh ``user@host`` for host-side execution, or ``""`` when no host is
    configured (mirror ``self_deploy``/``image_freshness._ssh_target``). On the prod
    host ``ORCH_DEPLOY_SSH_HOST=127.0.0.1`` is set by compose; the config default is
    empty so tests / non-self contexts fall back to the in-container command."""
    host = (settings.deploy_ssh_host or "").strip()
    if not host:
        return ""
    user = (settings.deploy_ssh_user or "").strip()
    return f"{user}@{host}" if user else host


def _channel_ssh_configured() -> bool:
    """Whether the execution channel is viable w.r.t. ssh config (fed to
    ``classify_staging_outcome``). In-container mode (host-side disabled) is always
    'configured' — its viability is decided purely by the result signals; host-side
    mode needs a non-empty ssh target (R-6: an empty target -> ``permanent-env``).
    never-raise -> assume configured (rely on the result signals)."""
    try:
        if not bool(getattr(settings, "staging_runner_exec_host_side", True)):
            return True
        return bool(_ssh_target())
    except Exception:  # noqa: BLE001 - never-raise
        return True


def build_staging_command() -> list[str]:
    """Build the staging-suite argv (same suite command the LLM-deployer ran).

    The INNER command is unchanged: ``docker exec <STAGING_SERVICE> python3
    <repos_dir>/<self-repo>/scripts/staging_check.py --base-url
    http://localhost:<staging_port> --mode stub``. Host-specifics come from config
    (ORCH-101, no host hardcodes).

    ORCH-123 (D1, adr-0049): ``docker`` lives on the HOST — the prod container ships
    only ``openssh-client git curl`` (``Dockerfile:11``), so a ``docker exec`` spawned
    FROM INSIDE the prod container hit ``FileNotFoundError`` (incident ORCH-116). When
    ``staging_runner_exec_host_side`` (default True) is set AND an ssh target is
    configured, the inner command is wrapped in ``ssh <user@host> '<docker exec ...>'``
    so it runs host-side over the existing trusted ssh channel (mirror
    ``self_deploy.build_deploy_command`` / ``image_freshness.image_revision``). With
    the flag off OR no ssh target configured (tests / non-self contexts) it falls back
    to the prior in-container ``docker exec`` (valid only where a docker CLI is baked
    into the image).

    Self-hosting safety (BR-7 / AC-8 / TC-08): the argv carries ONLY ``docker exec
    <staging-service> python3 staging_check.py ... --mode stub`` — NO restart of 8500,
    NO ``docker compose up orchestrator`` / ``--build``, NO force-push, NO ``.env``
    edit. The runner only reads/executes the staging suite (8501) and writes a log.
    """
    from .qg.checks import SELF_HOSTING_REPO
    repos_dir = (settings.repos_dir or "/repos").rstrip("/")
    script = f"{repos_dir}/{SELF_HOSTING_REPO}/scripts/staging_check.py"
    base_url = f"http://localhost:{int(settings.staging_port)}"
    inner_argv = [
        "docker", "exec", STAGING_SERVICE,
        "python3", script,
        "--base-url", base_url,
        "--mode", "stub",
    ]
    # ORCH-123 (D1): host-side ssh-wrap when enabled AND a target is configured.
    if bool(getattr(settings, "staging_runner_exec_host_side", True)):
        target = _ssh_target()
        if target:
            remote = " ".join(shlex.quote(a) for a in inner_argv)
            return ["ssh", "-o", "StrictHostKeyChecking=no", target, remote]
    # Fallback: prior in-container docker exec (no ssh target / flag off).
    return inner_argv


def _resolve_timeout() -> int:
    """Resolve ``staging_runner_timeout_s`` (malformed/non-positive -> default +
    WARNING, never-break — mirror of ``merge_gate._resolve_retest_timeout``)."""
    raw = getattr(settings, "staging_runner_timeout_s", _DEFAULT_TIMEOUT_S)
    try:
        t = int(raw)
        if t > 0:
            return t
        logger.warning(
            "staging_runner_timeout_s non-positive (%r) -> default %ds",
            raw, _DEFAULT_TIMEOUT_S,
        )
    except (TypeError, ValueError):
        logger.warning(
            "staging_runner_timeout_s malformed (%r) -> default %ds",
            raw, _DEFAULT_TIMEOUT_S,
        )
    return _DEFAULT_TIMEOUT_S


def run_staging_suite() -> proc_group.ProcResult:
    """Execute the staging suite in its own process group, tree-killed on timeout.

    Reuses ``proc_group.run_in_process_group`` (ORCH-110) so a hung docker-exec /
    pytest subtree is killed whole (no orphans, AC-9). Never raises (proc_group
    degrades any OS error to a safe ``ProcResult``).
    """
    cmd = build_staging_command()
    timeout = _resolve_timeout()
    try:
        grace = float(getattr(settings, "agent_kill_grace_seconds", 5) or 5)
    except (TypeError, ValueError):
        grace = 5.0
    return proc_group.run_in_process_group(
        cmd,
        cwd=None,
        timeout=timeout,
        grace_s=grace,
        tree_kill=bool(getattr(settings, "subprocess_tree_kill_enabled", True)),
    )


# ---------------------------------------------------------------------------
# exit-code -> verdict (D4 / FR-3 / AC-3) — reuse the single contract, no 2nd mapping
# ---------------------------------------------------------------------------
def map_exit_code_to_status(exit_code) -> str:
    """``0 -> SUCCESS``; non-zero / None / non-int -> ``FAILED`` (fail-closed).

    Delegates to ``self_deploy.map_exit_code_to_status`` — the SAME pure, unit-tested
    contract the deploy-finalizer uses (BR-4: no second, drifting mapping).
    """
    from .self_deploy import map_exit_code_to_status as _map
    return _map(exit_code)


# ---------------------------------------------------------------------------
# Three-way outcome classification (D3 / FR-2 / FR-3 / AC-3, AC-4, AC-6)
# ---------------------------------------------------------------------------
def classify_staging_outcome(result, ssh_configured: bool) -> str:
    """Classify a staging-suite ProcResult into one of three classes (D3). Pure,
    never-raise (mirror of ``merge_gate.classify_retest_failure``, ORCH-110 D2):

      * ``"suite-ran"``       — a recognised executor exit-code (any int except the
                                ssh transport code 255) AND no environment marker in
                                stderr: the suite DEMONSTRABLY executed -> trust the
                                code (``0->SUCCESS``, ``!=0->FAILED``). Anti-over-
                                tolerance (BR-3): a real suite fail is NEVER
                                reclassified as infra/env.
      * ``"permanent-env"``   — a deterministic PERMANENT environment defect: an
                                stderr env-marker (no docker / no such container /
                                daemon unreachable), a shell "command not found /
                                cannot execute" code (126/127), no host-side ssh
                                target configured, or a bare local spawn-error
                                (``returncode is None`` and not a timeout). Retrying is
                                pointless (FR-3) -> immediate distinguishable
                                infra-HOLD (no rollback, no developer-retry).
      * ``"transient-infra"`` — a timeout OR an ssh transport/connection failure (255)
                                OR any unknown signal: a retry is meaningful -> the
                                bounded DEFER.

    The exit=1 collision (``docker exec`` "No such container"=1 vs a real suite fail=1)
    is disambiguated by SCANNING stderr for env-markers, NEVER by the bare exit-code.
    Fail-safe on doubt -> ``"transient-infra"`` (DEFER), never a silent ``"suite-ran"``
    (a mis-routed environment->code-fail is exactly the defect this fixes; an over-
    tolerated code-fail is guarded by trusting recognised suite exit-codes first).
    """
    try:
        rc = getattr(result, "returncode", None)
        timed_out = bool(getattr(result, "timed_out", False))
        stderr = (getattr(result, "stderr", "") or "").lower()

        # 1. env-marker in stderr -> permanent, REGARDLESS of the exit-code (this is
        #    what disambiguates a `docker exec` "No such container" exit=1 from a real
        #    suite fail=1 — R-3).
        if any(m in stderr for m in _ENV_MARKERS):
            return "permanent-env"
        # 2. shell "command not found" (127) / "cannot execute" (126) -> permanent.
        if rc in (126, 127):
            return "permanent-env"
        # 3. a recognised executor exit-code (any int except the ssh transport code
        #    255) WITHOUT an env-marker -> the suite executed; trust it (BR-3). This is
        #    checked BEFORE the channel guards so a real suite fail is never masked.
        if isinstance(rc, int) and rc != 255:
            return "suite-ran"
        # --- below this line the suite did NOT produce a trustworthy exit-code ---
        # 4. timeout / ssh transport failure (255) -> transient: a retry is meaningful.
        if timed_out or rc == 255:
            return "transient-infra"
        # 5. host-side ssh target missing (R-6) OR a bare local spawn-error (proc_group
        #    degraded an OSError to rc None without a timeout) -> permanent env defect.
        if not ssh_configured:
            return "permanent-env"
        if rc is None:
            return "permanent-env"
        # 6. unknown signal -> fail-safe DEFER (never a silent suite-ran).
        return "transient-infra"
    except Exception:  # noqa: BLE001 - never-raise; unknown -> transient (no false rollback)
        return "transient-infra"


# ---------------------------------------------------------------------------
# Artifact 15-staging-log.md (D6 / FR-4 / AC-2 / AC-8) — mirror write_deploy_log
# ---------------------------------------------------------------------------
def _extract_infra_waived(stdout: str) -> list[str]:
    """Lines from the suite stdout carrying the ORCH-061 ``INFRA-WAIVED:`` marker
    (copied into the log body for observability, as the prompt required)."""
    if not stdout:
        return []
    return [ln.strip() for ln in stdout.splitlines() if "INFRA-WAIVED" in ln]


def build_staging_log(
    work_item_id: str, exit_code, status: str, stdout: str = "", *, tool_error: bool = False
) -> str:
    """Render a ``15-staging-log.md`` body whose ``staging_status:`` frontmatter is the
    verdict ``check_staging_status`` / ``_parse_staging_status`` reads (contract
    UNCHANGED, AC-2). Carries the mandatory 52c schema (``work_item`` / ``stage`` /
    ``author_agent`` / ``status`` / ``created_at`` / ``model_used``); ``author_agent:
    staging-runner`` / ``model_used: n/a`` honestly reflect the DETERMINISTIC producer.
    The machine key ``staging_status:`` and its UPPERCASE value are NOT changed.

    Written as a literal block (mirror of ``self_deploy.build_deploy_log``) so the
    machine-read frontmatter is byte-exact; only the frontmatter is machine-read, the
    body is informational.
    """
    import datetime
    created = datetime.date.today().isoformat()
    sub_status = "success" if status == "SUCCESS" else "failed"
    base_url = f"http://localhost:{int(settings.staging_port)}"

    waived = _extract_infra_waived(stdout)
    waived_body = ""
    if waived:
        waived_body = "\nINFRA-WAIVED lines (ORCH-061, copied for observability):\n" + "\n".join(
            f"- {ln}" for ln in waived
        ) + "\n"

    tail = ""
    if stdout:
        tail_text = stdout.strip()[-1500:]
        if tail_text:
            tail = f"\nStaging suite stdout (tail):\n```\n{tail_text}\n```\n"

    note = (
        "Staging suite did NOT execute (tool-error) and the infra-retry budget was "
        "exhausted -> fail-closed FAILED."
        if tool_error
        else f"Staging suite exit-code `{exit_code}` -> `staging_status: {status}`."
    )

    return (
        "---\n"
        f"staging_status: {status}\n"
        f"work_item: {work_item_id}\n"
        "stage: deploy-staging\n"
        "author_agent: staging-runner\n"
        f"status: {sub_status}\n"
        f"created_at: {created}\n"
        "model_used: n/a\n"
        f"exit_code: {exit_code}\n"
        f"base_url: {base_url}\n"
        "---\n\n"
        "# Staging Gate Log (deterministic runner, ORCH-115)\n\n"
        f"{note}\n\n"
        "Вердикт зафиксирован детерминированным staging-раннером (ORCH-115), не LLM. "
        "infra-tolerance (ORCH-061) уже учтена внутри `staging_check.py` — раннер её не "
        "пересуживает.\n"
        f"{waived_body}"
        f"{tail}"
    )


def write_staging_log(
    repo: str, work_item_id: str, branch: str, exit_code, status: str,
    stdout: str = "", *, tool_error: bool = False,
) -> bool:
    """Write ``15-staging-log.md`` into the task worktree (so ``check_staging_status``
    reads it) and best-effort commit+push it to the FEATURE BRANCH. Returns True iff
    the file was written. Never raises.

    Mirror of ``self_deploy.write_deploy_log`` EXCEPT: the actor name is
    ``staging-runner`` and the log is pushed only to the feature branch — there is NO
    separate PR-merge of the log into ``main`` (the gate reads the worktree first;
    excluding any direct work on ``main`` strengthens AC-8 / BR-7). The feature branch
    is merged into ``main`` later by the normal merge-gate path.
    """
    import os
    import subprocess
    from .git_worktree import get_worktree_path

    rel = f"docs/work-items/{work_item_id}/15-staging-log.md"
    try:
        wt = get_worktree_path(repo, branch)
    except Exception as e:  # noqa: BLE001 - never-raise
        logger.error("write_staging_log: worktree error for %s/%s: %s", repo, branch, e)
        return False

    path = os.path.join(wt, rel)
    content = build_staging_log(work_item_id, exit_code, status, stdout, tool_error=tool_error)
    try:
        os.makedirs(os.path.dirname(path), exist_ok=True)
        with open(path, "w", encoding="utf-8") as f:
            f.write(content)
    except OSError as e:
        logger.error("write_staging_log: write error at %s: %s", path, e)
        return False

    # Best-effort commit + push to the feature branch (the gate also falls back to
    # origin/main). ORCH-101: HOME + email domain from Settings; the actor NAME stays
    # the platform literal `staging-runner` (deterministic system-actor commits).
    _email = f"staging-runner@{settings.git_email_domain}"
    git_env = {
        **os.environ,
        "HOME": settings.agent_home_dir,
        "GIT_AUTHOR_NAME": "staging-runner",
        "GIT_AUTHOR_EMAIL": _email,
        "GIT_COMMITTER_NAME": "staging-runner",
        "GIT_COMMITTER_EMAIL": _email,
    }
    try:
        subprocess.run(["git", "-C", wt, "add", rel],
                       capture_output=True, timeout=_GIT_TIMEOUT, env=git_env)
        commit = subprocess.run(
            ["git", "-C", wt, "commit", "-m",
             f"staging(ORCH-115): staging gate {status} for {work_item_id}"],
            capture_output=True, text=True, timeout=_GIT_TIMEOUT, env=git_env,
        )
        if commit.returncode == 0:
            subprocess.run(["git", "-C", wt, "push", "origin", branch],
                           capture_output=True, timeout=_GIT_TIMEOUT, env=git_env)
    except (subprocess.SubprocessError, OSError) as e:
        logger.warning("write_staging_log: git commit/push best-effort failed: %s", e)
    return True


# ---------------------------------------------------------------------------
# Existing-gate initiation (D7 / FR-5 / AC-4) — no new routing branch
# ---------------------------------------------------------------------------
def _advance(task_id, repo: str, work_item_id: str, branch: str) -> None:
    """Initiate the SAME ``advance_stage`` evaluation a finished LLM-deployer would
    (``finished_agent="deployer"`` on ``deploy-staging``): SUCCESS -> sub-gates
    security->merge->coverage->image-freshness (ORCH-022/043/027/058) + Phase A
    (ORCH-036); FAILED -> the existing rollback deploy-staging -> development
    (``stage_engine.py:932``). No new routing branch. The transition-lease (ORCH-114)
    is taken INSIDE advance_stage on the side-effectful edge — the runner never
    touches it (task boundary O1). never-raise."""
    try:
        from . import stage_engine
        stage_engine.advance_stage(
            task_id=task_id,
            current_stage="deploy-staging",
            repo=repo,
            work_item_id=work_item_id,
            branch=branch,
            finished_agent="deployer",
        )
    except Exception as e:  # noqa: BLE001 - never-raise into the worker
        logger.error(
            "staging_runner._advance: advance_stage failed for task %s (%s): %s",
            task_id, work_item_id, e,
        )


# ---------------------------------------------------------------------------
# Two-level outcome (D5) — tool-error DEFER bookkeeping
# ---------------------------------------------------------------------------
def _infra_retry_count(task_id) -> int:
    """How many times this task was re-queued by the tool-error DEFER path
    (restart-safe; counted from the persisted jobs queue by the marker — mirror of
    ``stage_engine._merge_infra_retry_count``). Never raises -> 0 on error."""
    try:
        from .db import get_db
        conn = get_db()
        n = conn.execute(
            "SELECT COUNT(*) FROM jobs WHERE task_id=? AND task_content LIKE ?",
            (task_id, f"%{_INFRA_RETRY_MARKER}%"),
        ).fetchone()[0]
        conn.close()
        return int(n)
    except Exception as e:  # noqa: BLE001 - never-raise
        logger.warning("staging_runner._infra_retry_count error for %s: %s", task_id, e)
        return 0


def _handle_transient_infra(
    task_id, repo: str, work_item_id: str, branch: str, result: proc_group.ProcResult
) -> None:
    """Transient infra (timeout / ssh transport) — the suite did not execute but a
    retry is meaningful (D4). Bounded DEFER: re-queue a fresh ``deployer`` job (which
    re-enters this runner) with a delay + a restart-safe marker, instead of an
    immediate rollback that would burn a developer-retry.

    On budget exhaustion -> **infra-HOLD + alert** (D4, SUPERSEDES ORCH-115 D5): NOT
    the prior fail-closed ``write_staging_log("FAILED") + advance``, which falsely
    rolled an unresolved infra hiccup back to ``development`` as a code-fail (BR-2,
    anti-pattern ORCH-110). The task is HELD on ``deploy-staging`` (a red/missing gate
    keeps the reconciler's ``advance_if_gate_passed`` from rolling it back — R-2); the
    operator re-drives it after fixing the stand. Never a silent advance / false green;
    never wedges the queue. never-raise."""
    retries = _infra_retry_count(task_id)
    try:
        max_retries = int(settings.staging_runner_infra_max_retries)
    except (TypeError, ValueError):
        max_retries = 2
    try:
        delay = int(settings.staging_runner_infra_retry_delay_s)
    except (TypeError, ValueError):
        delay = 30

    if retries < max_retries:
        _bump("deferred")
        reason = "timeout" if result.timed_out else "ssh transport/connection error"
        task_desc = (
            f"Work item: {work_item_id}\nRepo: {repo}\nBranch: {branch}\n"
            f"Stage: deploy-staging\nNote: {_INFRA_RETRY_MARKER} "
            f"(attempt {retries + 1}/{max_retries}) — {reason}, retrying after {delay}s."
        )
        try:
            from .db import enqueue_job
            new_job = enqueue_job(
                "deployer", repo, task_desc, task_id=task_id, available_at_delay_s=delay,
            )
            logger.warning(
                "Task %s (%s): staging suite did not execute (%s) -> transient-infra "
                "DEFER (job_id=%s, attempt %d/%d)",
                task_id, work_item_id, reason, new_job, retries + 1, max_retries,
            )
        except Exception as e:  # noqa: BLE001 - never-raise
            logger.error("staging_runner: infra-DEFER enqueue failed for %s: %s", task_id, e)
        return

    # Budget exhausted -> infra-HOLD (D4): NO write_staging_log("FAILED"), NO advance,
    # NO developer-retry. The task is held on deploy-staging for the operator.
    _bump("permanent_env")
    logger.error(
        "Task %s (%s): staging transient-infra DEFER budget exhausted (%d) -> "
        "infra-HOLD on deploy-staging (NOT a code defect, no rollback)",
        task_id, work_item_id, max_retries,
    )
    _alert_infra_exhausted(work_item_id, max_retries)


def _handle_permanent_env(
    task_id, repo: str, work_item_id: str, branch: str, result: proc_group.ProcResult
) -> None:
    """Permanent environment defect (D4): an immediate, distinguishable infra-HOLD.

    The DEFER cycle is SKIPPED — retrying a permanent defect (no docker / no ssh
    target / container down) is pointless (FR-3). We do NOT write a FAILED staging-log,
    do NOT advance, do NOT burn a developer-retry — the task is HELD on
    ``deploy-staging``: a red/missing ``check_staging_status`` keeps the reconciler's
    ``advance_if_gate_passed`` from rolling it back to ``development`` (R-2), and the
    rollback-to-development path (``advance_stage(finished_agent="deployer")``) is never
    taken. A structured error log + an operator alert ("infra/environment, NOT a code
    defect") make the hold visible; the operator re-drives after fixing the
    environment. never-raise."""
    detail = (getattr(result, "stderr", "") or "").strip()[:300]
    logger.error(
        "Task %s (%s): PERMANENT staging environment defect -> infra-HOLD on "
        "deploy-staging (NOT a code defect, no rollback, no developer-retry): %s",
        task_id, work_item_id, detail or "<no stderr>",
    )
    _alert_permanent_env(work_item_id, detail)


def _alert_infra_exhausted(work_item_id: str, max_retries: int) -> None:
    """Best-effort Telegram alert that the staging suite never executed (infra, NOT a
    code defect) after the retry budget -> infra-HOLD (no rollback). never-raise."""
    try:
        from .notifications import send_telegram, link_for
        send_telegram(
            f"\U0001f6a8 {link_for(work_item_id)}: staging-сюита не запустилась "
            f"(инфра, НЕ дефект кода) после {max_retries} попыток — задача удержана на "
            f"deploy-staging (без отката на development). Нужно проверить staging-стенд."
        )
    except Exception as e:  # noqa: BLE001 - never-raise
        logger.warning("staging_runner: infra-exhausted alert failed for %s: %s", work_item_id, e)


def _alert_permanent_env(work_item_id: str, detail: str) -> None:
    """Best-effort Telegram alert that the staging suite could not execute due to a
    permanent environment defect (NOT a code defect) -> infra-HOLD. never-raise."""
    try:
        from .notifications import send_telegram, link_for
        tail = f": {detail}" if detail else ""
        send_telegram(
            f"\U0001f6a8 {link_for(work_item_id)}: staging-сюита не смогла исполниться — "
            f"постоянный дефект окружения (инфра, НЕ дефект кода){tail}. Задача удержана "
            f"на deploy-staging до починки окружения (без отката на development)."
        )
    except Exception as e:  # noqa: BLE001 - never-raise
        logger.warning("staging_runner: permanent-env alert failed for %s: %s", work_item_id, e)


# ---------------------------------------------------------------------------
# Entry point (D2) — owns the full deterministic flow, mirror run_deploy_finalizer
# ---------------------------------------------------------------------------
def run_staging_gate(job: dict) -> None:
    """Deterministic staging gate for a ``deployer`` job on ``deploy-staging``.

    Flow (mirror of ``stage_engine.run_deploy_finalizer``):
      1. resolve ``work_item_id`` / ``branch`` by ``task_id``;
      2. execute the staging suite (D3) -> ProcResult;
      3. classify the outcome three ways (D3): ``suite-ran`` -> map exit-code ->
         ``staging_status:``, write ``15-staging-log.md``, initiate the existing gate
         via ``advance_stage`` (D7);
      4. ``permanent-env`` -> immediate infra-HOLD (no rollback, no developer-retry);
         ``transient-infra`` -> bounded DEFER, then infra-HOLD on exhaustion (D4);
      5. observability counters + one structured verdict log (D8).
    Never raises into the caller (the launcher marks the job done/failed)."""
    started = time.time()
    _bump("runs")
    task_id = job.get("task_id")
    repo = job.get("repo")

    # 1. resolve task fields.
    work_item_id, branch = None, None
    try:
        from .db import get_db
        conn = get_db()
        row = conn.execute(
            "SELECT work_item_id, branch FROM tasks WHERE id=?", (task_id,)
        ).fetchone()
        conn.close()
        if row:
            work_item_id, branch = row[0], row[1]
    except Exception as e:  # noqa: BLE001 - never-raise
        logger.error("staging_runner: task lookup failed for task_id=%s: %s", task_id, e)
    if not work_item_id or not branch:
        logger.error(
            "staging_runner: missing work_item_id/branch for task_id=%s — aborting", task_id
        )
        return

    # 2-4. execute + classify + route — guarded so AC-7 (never-raise) holds even if
    # an unexpected error escapes a sub-step (the worker must never crash on staging
    # infra; the task is left on deploy-staging for the reconciler/reaper to re-drive).
    try:
        result = run_staging_suite()
        duration_s = round(time.time() - started, 1)
        # ORCH-123 (D3): three-way classification (env != transient infra != code-fail).
        outcome = classify_staging_outcome(result, _channel_ssh_configured())

        if outcome == "suite-ran":
            # 3. trust the exit-code (ORCH-061 already inside staging_check.py).
            status = map_exit_code_to_status(result.returncode)
            _bump("success" if status == "SUCCESS" else "failed")
            logger.info(
                "staging_runner verdict: work_item=%s repo=%s exit_code=%s status=%s "
                "duration_s=%s outcome=%s",
                work_item_id, repo, result.returncode, status, duration_s,
                "code-pass" if status == "SUCCESS" else "code-fail",
            )
            write_staging_log(repo, work_item_id, branch, result.returncode, status, result.stdout)
            _advance(task_id, repo, work_item_id, branch)
            return

        # The suite did NOT execute. Count it as a tool-error, then route by class
        # (D4): permanent-env -> immediate infra-HOLD; transient-infra -> bounded DEFER.
        # NEITHER rolls back to development or burns a developer-retry (BR-2 invariant).
        _bump("tool_error")
        if outcome == "permanent-env":
            _bump("permanent_env")
            logger.warning(
                "staging_runner verdict: work_item=%s repo=%s exit_code=%s "
                "duration_s=%s outcome=permanent-env (ssh_configured=%s)",
                work_item_id, repo, result.returncode, duration_s, _channel_ssh_configured(),
            )
            _handle_permanent_env(task_id, repo, work_item_id, branch, result)
            return

        # transient-infra (timeout / ssh transport / unknown) -> bounded DEFER.
        logger.warning(
            "staging_runner verdict: work_item=%s repo=%s exit_code=%s "
            "duration_s=%s outcome=transient-infra (timed_out=%s)",
            work_item_id, repo, result.returncode, duration_s, result.timed_out,
        )
        _handle_transient_infra(task_id, repo, work_item_id, branch, result)
    except Exception as e:  # noqa: BLE001 - never-raise into the worker (AC-7)
        logger.error(
            "staging_runner.run_staging_gate: unexpected error for task %s (%s): %s",
            task_id, work_item_id, e,
        )


# ---------------------------------------------------------------------------
# Prod-like preflight of the host-side execution channel (D5 / FR-4 / AC-5)
# ---------------------------------------------------------------------------
def _alert_preflight(reason: str) -> None:
    """Best-effort Telegram alert that the host-side staging channel is unworkable at
    startup (so a real task never silently false-routes later). never-raise."""
    try:
        from .notifications import send_telegram
        send_telegram(
            f"⚠️ staging-runner preflight: {reason}. Хост-сторона исполнения "
            f"staging-сюиты неработоспособна — задачи на deploy-staging будут удержаны "
            f"(инфра, НЕ дефект кода). Проверьте ssh/docker/staging-стенд."
        )
    except Exception as e:  # noqa: BLE001 - never-raise
        logger.warning("staging_runner: preflight alert failed: %s", e)


def preflight() -> tuple[bool, str]:
    """Prod-like preflight of the host-side staging execution channel (D5 / AC-5).

    Probes the channel with a short bounded ssh probe (``command -v docker`` +
    ``docker inspect -f '{{.State.Running}}' <staging-service>``) so a broken
    environment (no docker on the host / staging container down / ssh unreachable / no
    ssh target configured) surfaces at SERVICE START — NOT postfactum as a false
    rollback of a real task. Records ``_PREFLIGHT_STATE`` + alerts on failure.

    Purely observational (FR-4): it NEVER gates/blocks the pipeline, touches stages /
    QG, or raises — called best-effort from ``main.lifespan``. Self-hosting scope:
    ``applies()`` first (a disabled runner / out-of-scope repo -> n/a). Returns
    ``(ok, reason)`` (``ok=True`` for n/a / in-container mode / a healthy channel).
    """
    try:
        from .qg.checks import SELF_HOSTING_REPO
        if not applies(SELF_HOSTING_REPO):
            _PREFLIGHT_STATE.update(ok=None, reason="n/a (runner disabled / out of scope)")
            return True, "n/a"
        # In-container mode (host-side disabled): nothing to probe host-side.
        if not bool(getattr(settings, "staging_runner_exec_host_side", True)):
            _PREFLIGHT_STATE.update(ok=True, reason="in-container mode (host-side disabled)")
            return True, "in-container mode"
        target = _ssh_target()
        if not target:
            reason = "no ssh target configured (deploy_ssh_host empty) — host-side staging unworkable"
            _PREFLIGHT_STATE.update(ok=False, reason=reason)
            _alert_preflight(reason)
            return False, reason
        probe = (
            "command -v docker >/dev/null 2>&1 && "
            f"docker inspect -f '{{{{.State.Running}}}}' {shlex.quote(STAGING_SERVICE)}"
        )
        cmd = ["ssh", "-o", "StrictHostKeyChecking=no", target, probe]
        try:
            r = subprocess.run(cmd, capture_output=True, text=True, timeout=_PREFLIGHT_TIMEOUT_S)
        except subprocess.TimeoutExpired:
            reason = "preflight ssh probe timed out"
            _PREFLIGHT_STATE.update(ok=False, reason=reason)
            _alert_preflight(reason)
            return False, reason
        except (subprocess.SubprocessError, OSError) as e:
            reason = f"preflight ssh probe error: {e}"
            _PREFLIGHT_STATE.update(ok=False, reason=reason)
            _alert_preflight(reason)
            return False, reason
        out = (r.stdout or "").strip().lower()
        if r.returncode == 0 and out == "true":
            _PREFLIGHT_STATE.update(ok=True, reason="host-side channel ok (docker present, staging running)")
            return True, "ok"
        reason = f"host-side staging channel not ready (rc={r.returncode}, running={out!r})"
        _PREFLIGHT_STATE.update(ok=False, reason=reason)
        _alert_preflight(reason)
        return False, reason
    except Exception as e:  # noqa: BLE001 - never-raise; preflight must never block start
        logger.warning("staging_runner.preflight error: %s", e)
        _PREFLIGHT_STATE.update(ok=None, reason=f"preflight error: {e}")
        return True, "preflight skipped (error)"


# ---------------------------------------------------------------------------
# Observability (D8 / FR-7 / AC-10)
# ---------------------------------------------------------------------------
def snapshot() -> dict:
    """Read-only staging-runner summary for ``GET /queue`` (FR-7 / AC-10).

    Additive block; existing ``/queue`` keys are untouched. ORCH-123 adds
    ``exec_host_side`` (the strategy flag), ``permanent_env`` (the infra-HOLD counter,
    distinct from ``failed``=code-fail and ``deferred``=transient-infra) and the last
    ``preflight`` verdict. never-raise: any error -> a minimal dict with the
    kill-switch state."""
    try:
        return {
            "enabled": bool(settings.staging_runner_enabled),
            "repos": getattr(settings, "staging_runner_repos", "") or "",
            "timeout_s": getattr(settings, "staging_runner_timeout_s", _DEFAULT_TIMEOUT_S),
            "infra_max_retries": getattr(settings, "staging_runner_infra_max_retries", 2),
            "exec_host_side": bool(getattr(settings, "staging_runner_exec_host_side", True)),
            "runs": _STAGING_RUNNER_COUNTERS["runs"],
            "success": _STAGING_RUNNER_COUNTERS["success"],
            "failed": _STAGING_RUNNER_COUNTERS["failed"],
            "tool_error": _STAGING_RUNNER_COUNTERS["tool_error"],
            "deferred": _STAGING_RUNNER_COUNTERS["deferred"],
            "permanent_env": _STAGING_RUNNER_COUNTERS["permanent_env"],
            "preflight": dict(_PREFLIGHT_STATE),
        }
    except Exception as e:  # noqa: BLE001 - never-raise -> minimal dict
        logger.warning("staging_runner.snapshot error: %s", e)
        return {"enabled": False}
