"""ORCH-123 — staging-runner execution strategy must not depend on a docker CLI
inside the prod app-container (host-side ssh + three-way env classification).

Background (incident ORCH-116): the ORCH-115 deterministic staging-runner ran the
staging suite via ``docker exec`` FROM INSIDE the prod ``orchestrator`` container,
which ships only ``openssh-client git curl`` — NOT a ``docker`` CLI (Dockerfile:11).
``Popen(["docker", ...])`` hit ``FileNotFoundError`` -> a PERMANENT environment defect
that ORCH-115 mis-routed as a code-fail rollback ``deploy-staging -> development``
(burning developer-retries). The fix (ADR-001 / adr-0049):

  * D1 — execute the suite HOST-SIDE over the existing trusted ssh channel
    (ORCH-036/058), wrapping the SAME ``docker exec ...`` command;
  * D3 — three-way classification ``suite-ran`` / ``permanent-env`` / ``transient-infra``;
  * D4 — environment/transient-infra NEVER ends in a code-fail rollback (infra-HOLD);
    a really-executed failing suite STILL rolls back (anti-over-tolerance);
  * D5 — a prod-like preflight of the host-side channel at startup.

The pipeline gate / artifact contract / STAGE_TRANSITIONS / DB schema are byte-for-byte
unchanged (NFR-1). No live ssh / docker / network: the suite subprocess, advance_stage
and notifications are mocked; spawn-errors are reproduced with a non-existent binary.
"""
import os
import tempfile

import pytest

_test_db = os.path.join(tempfile.gettempdir(), "test_orch123_staging_runner_exec.db")
os.environ["ORCH_DB_PATH"] = _test_db
os.environ["ORCH_REPOS_DIR"] = tempfile.gettempdir()
os.environ.setdefault("ORCH_GITEA_TOKEN", "test-token")
os.environ.setdefault("ORCH_PLANE_API_TOKEN", "test-token")

from unittest.mock import MagicMock  # noqa: E402

import src.db as _db  # noqa: E402
from src.db import init_db, get_db  # noqa: E402
from src import staging_runner  # noqa: E402
from src import stage_engine  # noqa: E402
from src import config as cfg  # noqa: E402
from src.proc_group import ProcResult  # noqa: E402


# ---------------------------------------------------------------------------
# Fixtures (mirror tests/test_orch115_staging_runner.py)
# ---------------------------------------------------------------------------
@pytest.fixture(autouse=True)
def fresh_db(monkeypatch, tmp_path):
    monkeypatch.setattr(_db.settings, "db_path", _test_db)
    if os.path.exists(_test_db):
        os.unlink(_test_db)
    init_db()
    monkeypatch.setattr("src.git_worktree.settings.worktrees_dir", str(tmp_path), raising=False)
    # Isolate repos_dir too: check_staging_status falls back to <repos_dir>/<repo>
    # (and origin/main on it) when the feature worktree is absent. Without this the
    # gate would read REAL on-disk artifacts from the shared clone (e.g. a merged
    # ORCH-123/15-staging-log.md), turning the intended-RED gate in test_r2 green and
    # making the suite order-dependent. Point it at an empty tmp subdir (no .git, no
    # work-items) so the staging gate is deterministically "not found" -> red.
    monkeypatch.setattr(cfg.settings, "repos_dir", str(tmp_path / "repos"), raising=False)
    # Runner ON, self-hosting scope, host-side strategy ON (defaults).
    monkeypatch.setattr(cfg.settings, "staging_runner_enabled", True, raising=False)
    monkeypatch.setattr(cfg.settings, "staging_runner_repos", "", raising=False)
    monkeypatch.setattr(cfg.settings, "staging_runner_infra_max_retries", 2, raising=False)
    monkeypatch.setattr(cfg.settings, "staging_runner_exec_host_side", True, raising=False)
    # A configured ssh target (prod compose sets ORCH_DEPLOY_SSH_HOST=127.0.0.1).
    monkeypatch.setattr(cfg.settings, "deploy_ssh_host", "127.0.0.1", raising=False)
    monkeypatch.setattr(cfg.settings, "deploy_ssh_user", "slin", raising=False)
    for k in staging_runner._STAGING_RUNNER_COUNTERS:
        staging_runner._STAGING_RUNNER_COUNTERS[k] = 0
    staging_runner._PREFLIGHT_STATE.update(ok=None, reason="not-probed")
    yield


def _make_task(stage, repo="orchestrator", wi="ORCH-123", branch="feature/ORCH-123-x"):
    conn = get_db()
    cur = conn.execute(
        "INSERT INTO tasks (plane_id, work_item_id, repo, branch, stage) VALUES (?, ?, ?, ?, ?)",
        (f"plane-{wi}", wi, repo, branch, stage),
    )
    tid = cur.lastrowid
    conn.commit()
    conn.close()
    return tid


def _job_dict(job_id, agent, repo, task_id):
    return {"id": job_id, "agent": agent, "repo": repo, "task_id": task_id, "task_content": "x"}


def _infra_jobs(task_id):
    conn = get_db()
    n = conn.execute(
        "SELECT COUNT(*) FROM jobs WHERE task_id=? AND task_content LIKE ?",
        (task_id, f"%{staging_runner._INFRA_RETRY_MARKER}%"),
    ).fetchone()[0]
    conn.close()
    return n


def _read_log(repo, branch, wi):
    from src.git_worktree import get_worktree_path
    p = os.path.join(get_worktree_path(repo, branch), f"docs/work-items/{wi}/15-staging-log.md")
    with open(p, "r", encoding="utf-8") as f:
        return f.read()


def _log_exists(repo, branch, wi):
    from src.git_worktree import get_worktree_path
    return os.path.exists(
        os.path.join(get_worktree_path(repo, branch), f"docs/work-items/{wi}/15-staging-log.md")
    )


# ---------------------------------------------------------------------------
# TC-01 — MANDATORY regression (red->green): no docker CLI in the container
# ---------------------------------------------------------------------------
def test_tc01_regression_no_docker_cli_in_container(monkeypatch):
    """Reproduces incident ORCH-116. PRE-FIX: an in-container ``docker exec`` where the
    container has no docker CLI -> proc_group spawn-error -> tool-error -> infra-DEFER x2
    -> false FAILED rollback to development. POST-FIX: classified ``permanent-env`` ->
    infra-HOLD: NO advance, NO re-queue, NO developer-retry. (Red on pre-fix code —
    there is no permanent-env class / counter there and the path advances.)"""
    tid = _make_task("deploy-staging")
    # Force the pre-fix in-container shape, then make the spawn fail exactly as if
    # `docker` were absent — no network / ssh / docker needed.
    monkeypatch.setattr(cfg.settings, "staging_runner_exec_host_side", False)
    monkeypatch.setattr(staging_runner, "build_staging_command",
                        lambda: ["orch123-no-such-binary", "exec", "orchestrator-staging"])
    advance = MagicMock()
    monkeypatch.setattr(stage_engine, "advance_stage", advance)

    staging_runner.run_staging_gate(_job_dict(1, "deployer", "orchestrator", tid))

    advance.assert_not_called()                      # NO false rollback to development
    assert _infra_jobs(tid) == 0                     # DEFER cycle skipped (FR-3)
    assert not _log_exists("orchestrator", "feature/ORCH-123-x", "ORCH-123")  # no FAILED log
    assert staging_runner._STAGING_RUNNER_COUNTERS["permanent_env"] == 1
    assert staging_runner._STAGING_RUNNER_COUNTERS["tool_error"] == 1
    assert stage_engine.developer_retry_count(tid) == 0  # developer-retry not burned


# ---------------------------------------------------------------------------
# TC-02 — strategy does not require a docker CLI on the container PATH
# ---------------------------------------------------------------------------
def test_tc02_host_side_command_does_not_assume_in_container_docker():
    # With host-side ON + an ssh target, the suite is dispatched via ssh — the prod
    # container's own PATH is never asked for a `docker` binary.
    cmd = staging_runner.build_staging_command()
    assert cmd[0] == "ssh"                           # dispatched over the trusted channel
    assert "slin@127.0.0.1" in cmd
    # The INNER command is the SAME staging suite (contract unchanged).
    remote = cmd[-1]
    assert "docker exec orchestrator-staging" in remote
    assert "staging_check.py" in remote
    assert "--mode stub" in remote
    # The local executable invoked is `ssh` (in the container image), NOT `docker`.
    assert cmd[0] != "docker"


# ---------------------------------------------------------------------------
# TC-03 — environment/tool-error does NOT cause a misleading code-fail rollback
# ---------------------------------------------------------------------------
def test_tc03_env_defect_no_code_fail_rollback_and_alerts(monkeypatch):
    tid = _make_task("deploy-staging")
    monkeypatch.setattr(
        staging_runner, "run_staging_suite",
        lambda: ProcResult(returncode=None, stdout="",
                           stderr="[Errno 2] No such file or directory: 'docker'", timed_out=False),
    )
    advance = MagicMock()
    monkeypatch.setattr(stage_engine, "advance_stage", advance)
    sent = []
    monkeypatch.setattr("src.notifications.send_telegram", lambda msg, **kw: sent.append(msg))

    staging_runner.run_staging_gate(_job_dict(1, "deployer", "orchestrator", tid))

    advance.assert_not_called()                       # not a code-fail -> no rollback
    assert _infra_jobs(tid) == 0                       # permanent defect -> no pointless retry
    assert stage_engine.developer_retry_count(tid) == 0
    # A distinguishable infra/environment alert ("NOT a code defect") was sent.
    assert sent and any("НЕ дефект кода" in m for m in sent)


# ---------------------------------------------------------------------------
# TC-04 — anti-over-tolerance: a really-executed failing suite STILL rolls back
# ---------------------------------------------------------------------------
def test_tc04_real_suite_failure_still_rolls_back(monkeypatch):
    tid = _make_task("deploy-staging")
    # The suite executed (a real exit-code) and failed; clean stderr (no env-marker).
    monkeypatch.setattr(
        staging_runner, "run_staging_suite",
        lambda: ProcResult(returncode=1, stdout="B7 FAIL: assertion", stderr="", timed_out=False),
    )
    advance = MagicMock()
    monkeypatch.setattr(stage_engine, "advance_stage", advance)

    staging_runner.run_staging_gate(_job_dict(1, "deployer", "orchestrator", tid))

    advance.assert_called_once()                       # same rollback path as ORCH-115
    kwargs = advance.call_args.kwargs
    assert kwargs["current_stage"] == "deploy-staging"
    assert kwargs["finished_agent"] == "deployer"
    assert "staging_status: FAILED" in _read_log("orchestrator", "feature/ORCH-123-x", "ORCH-123")
    assert staging_runner._STAGING_RUNNER_COUNTERS["failed"] == 1
    assert staging_runner._STAGING_RUNNER_COUNTERS["permanent_env"] == 0


def test_tc04_exit1_with_env_marker_is_not_a_code_fail(monkeypatch):
    # The exit=1 collision: `docker exec` "No such container"=1 must NOT be read as a
    # suite fail=1. The stderr env-marker disambiguates it -> permanent-env (no rollback).
    tid = _make_task("deploy-staging")
    monkeypatch.setattr(
        staging_runner, "run_staging_suite",
        lambda: ProcResult(returncode=1, stdout="",
                           stderr="Error: No such container: orchestrator-staging", timed_out=False),
    )
    advance = MagicMock()
    monkeypatch.setattr(stage_engine, "advance_stage", advance)
    monkeypatch.setattr("src.notifications.send_telegram", lambda *a, **kw: None)

    staging_runner.run_staging_gate(_job_dict(1, "deployer", "orchestrator", tid))

    advance.assert_not_called()
    assert staging_runner._STAGING_RUNNER_COUNTERS["permanent_env"] == 1


# ---------------------------------------------------------------------------
# TC-05 — happy-path: suite ran, exit 0 -> SUCCESS -> advance
# ---------------------------------------------------------------------------
def test_tc05_happy_path_success_advances(monkeypatch):
    tid = _make_task("deploy-staging")
    monkeypatch.setattr(
        staging_runner, "run_staging_suite",
        lambda: ProcResult(returncode=0, stdout="ALL OK", stderr="", timed_out=False),
    )
    advance = MagicMock()
    monkeypatch.setattr(stage_engine, "advance_stage", advance)

    staging_runner.run_staging_gate(_job_dict(1, "deployer", "orchestrator", tid))

    advance.assert_called_once()
    assert advance.call_args.kwargs["finished_agent"] == "deployer"
    assert "staging_status: SUCCESS" in _read_log("orchestrator", "feature/ORCH-123-x", "ORCH-123")
    assert staging_runner._STAGING_RUNNER_COUNTERS["success"] == 1


# ---------------------------------------------------------------------------
# TC-06 — three-way classification (D3)
# ---------------------------------------------------------------------------
def _pr(rc=None, stderr="", timed_out=False):
    return ProcResult(returncode=rc, stdout="", stderr=stderr, timed_out=timed_out)


def test_tc06_classification_three_way():
    C = staging_runner.classify_staging_outcome
    # suite-ran: any recognised int (except 255) with no env-marker -> trust it.
    assert C(_pr(rc=0), True) == "suite-ran"
    assert C(_pr(rc=1), True) == "suite-ran"            # anti-over-tolerance (BR-3)
    assert C(_pr(rc=2, stderr="B7 FAIL"), True) == "suite-ran"
    # permanent-env: spawn-error (rc None, not a timeout).
    assert C(_pr(rc=None), True) == "permanent-env"
    # permanent-env: shell command-not-found / cannot-execute codes.
    assert C(_pr(rc=127), True) == "permanent-env"
    assert C(_pr(rc=126), True) == "permanent-env"
    # permanent-env: env-marker regardless of the exit-code.
    assert C(_pr(rc=1, stderr="Cannot connect to the Docker daemon"), True) == "permanent-env"
    assert C(_pr(rc=1, stderr="docker: command not found"), True) == "permanent-env"
    # permanent-env: host-side ssh target not configured (R-6).
    assert C(_pr(rc=None), False) == "permanent-env"
    # transient-infra: timeout / ssh transport (255) -> retry is meaningful.
    assert C(_pr(timed_out=True), True) == "transient-infra"
    assert C(_pr(rc=255), True) == "transient-infra"


def test_tc06_classification_never_raises():
    class Boom:
        @property
        def returncode(self):
            raise RuntimeError("boom")
    # An exploding result -> fail-safe transient-infra (DEFER, never a silent suite-ran).
    assert staging_runner.classify_staging_outcome(Boom(), True) == "transient-infra"


def test_tc06_transient_infra_defers_not_rolls_back(monkeypatch):
    tid = _make_task("deploy-staging")
    monkeypatch.setattr(staging_runner, "run_staging_suite",
                        lambda: _pr(rc=255, stderr="ssh: connect: Connection refused"))
    advance = MagicMock()
    monkeypatch.setattr(stage_engine, "advance_stage", advance)

    staging_runner.run_staging_gate(_job_dict(1, "deployer", "orchestrator", tid))

    advance.assert_not_called()                        # transient -> DEFER, never rollback
    assert _infra_jobs(tid) == 1                        # a fresh deployer job re-queued
    assert staging_runner._STAGING_RUNNER_COUNTERS["deferred"] == 1


# ---------------------------------------------------------------------------
# TC-07 — prod-like preflight (D5 / AC-5)
# ---------------------------------------------------------------------------
def test_tc07_preflight_no_ssh_target_signals_before_a_real_task(monkeypatch):
    monkeypatch.setattr(cfg.settings, "deploy_ssh_host", "")     # host-side, but no target
    sent = []
    monkeypatch.setattr("src.notifications.send_telegram", lambda msg, **kw: sent.append(msg))

    ok, reason = staging_runner.preflight()

    assert ok is False
    assert "no ssh target" in reason
    assert staging_runner._PREFLIGHT_STATE["ok"] is False
    assert sent                                         # operator alerted at startup


def test_tc07_preflight_probe_success(monkeypatch):
    class _R:
        returncode = 0
        stdout = "true\n"
        stderr = ""
    monkeypatch.setattr(staging_runner.subprocess, "run", lambda *a, **kw: _R())
    ok, reason = staging_runner.preflight()
    assert ok is True
    assert staging_runner._PREFLIGHT_STATE["ok"] is True


def test_tc07_preflight_probe_failure_alerts(monkeypatch):
    class _R:
        returncode = 1
        stdout = ""           # docker missing on host / container down
        stderr = "command not found"
    monkeypatch.setattr(staging_runner.subprocess, "run", lambda *a, **kw: _R())
    sent = []
    monkeypatch.setattr("src.notifications.send_telegram", lambda msg, **kw: sent.append(msg))
    ok, _reason = staging_runner.preflight()
    assert ok is False
    assert sent


def test_tc07_preflight_noop_when_disabled(monkeypatch):
    monkeypatch.setattr(cfg.settings, "staging_runner_enabled", False)
    ok, reason = staging_runner.preflight()            # out of scope -> n/a, never blocks
    assert ok is True
    assert reason == "n/a"


def test_tc07_preflight_in_container_mode_noop(monkeypatch):
    monkeypatch.setattr(cfg.settings, "staging_runner_exec_host_side", False)
    ok, reason = staging_runner.preflight()
    assert ok is True
    assert "in-container" in reason


# ---------------------------------------------------------------------------
# TC-08 — self-hosting safety: no dangerous literals in the host-side command
# ---------------------------------------------------------------------------
def test_tc08_host_side_command_has_no_dangerous_literals():
    cmd = " ".join(staging_runner.build_staging_command())
    forbidden = ("compose", "up -d", "--build", "8500", "force", "push", ".env",
                 "rm ", "restart")
    for token in forbidden:
        assert token not in cmd, f"forbidden literal {token!r} in host-side command: {cmd}"
    assert "ssh" in cmd and "docker exec" in cmd and "staging_check.py" in cmd
    assert "8501" in cmd and "--mode stub" in cmd


# ---------------------------------------------------------------------------
# TC-09 — kill-switch / scope / strategy flag default (AC-10)
# ---------------------------------------------------------------------------
def test_tc09_killswitch_scope_and_flag_default(monkeypatch):
    # kill-switch off -> should_intercept False -> the prior LLM deployer via _spawn.
    monkeypatch.setattr(cfg.settings, "staging_runner_enabled", False)
    tid = _make_task("deploy-staging")
    assert staging_runner.should_intercept(_job_dict(1, "deployer", "orchestrator", tid)) is False
    # enabled, empty CSV -> self-hosting only.
    monkeypatch.setattr(cfg.settings, "staging_runner_enabled", True)
    assert staging_runner.applies("orchestrator") is True
    assert staging_runner.applies("enduro-trails") is False
    # the strategy flag defaults to host-side (prod).
    assert cfg.settings.staging_runner_exec_host_side is True


def test_tc09_host_side_off_falls_back_to_in_container(monkeypatch):
    monkeypatch.setattr(cfg.settings, "staging_runner_exec_host_side", False)
    cmd = staging_runner.build_staging_command()
    assert cmd[:2] == ["docker", "exec"]               # rollback knob -> prior 1:1 shape


def test_tc09_host_side_on_but_no_target_falls_back(monkeypatch):
    monkeypatch.setattr(cfg.settings, "deploy_ssh_host", "")
    cmd = staging_runner.build_staging_command()
    assert cmd[:2] == ["docker", "exec"]               # no target -> in-container fallback


# ---------------------------------------------------------------------------
# TC-10 — artifact contract: staging_status frontmatter + 52c schema (AC-7)
# ---------------------------------------------------------------------------
def test_tc10_artifact_contract_unchanged():
    body = staging_runner.build_staging_log("ORCH-123", 0, "SUCCESS", "ok")
    assert "staging_status: SUCCESS" in body            # UPPERCASE machine key, unchanged
    for field in ("work_item: ORCH-123", "stage: deploy-staging",
                  "author_agent: staging-runner", "model_used: n/a"):
        assert field in body
    # The UNCHANGED gate parser reads the runner's frontmatter.
    from src.qg.checks import _parse_staging_status
    ok, _r = _parse_staging_status(body)
    assert ok is True
    bad, _r2 = _parse_staging_status(staging_runner.build_staging_log("ORCH-123", 1, "FAILED"))
    assert bad is False


# ---------------------------------------------------------------------------
# TC-11 — pipeline anti-regress: gate / transitions / schema untouched (AC-7)
# ---------------------------------------------------------------------------
def test_tc11_pipeline_contract_unchanged():
    from src.stages import STAGE_TRANSITIONS
    from src.qg.checks import QG_CHECKS
    assert STAGE_TRANSITIONS["deploy-staging"] == {
        "next": "deploy", "agent": "deployer", "qg": "check_staging_status"
    }
    assert "check_staging_status" in QG_CHECKS
    conn = get_db()
    tables = {r[0] for r in conn.execute(
        "SELECT name FROM sqlite_master WHERE type='table'").fetchall()}
    conn.close()
    assert not any("staging_runner" in t for t in tables)   # no DB migration


# ---------------------------------------------------------------------------
# TC-12 — never-raise / queue not wedged + observability distinguishes classes
# ---------------------------------------------------------------------------
def test_tc12_run_gate_never_raises(monkeypatch):
    tid = _make_task("deploy-staging")

    def boom():
        raise RuntimeError("ssh exploded")
    monkeypatch.setattr(staging_runner, "run_staging_suite", boom)
    staging_runner.run_staging_gate(_job_dict(1, "deployer", "orchestrator", tid))  # must NOT raise


def test_tc12_snapshot_distinguishes_classes():
    snap = staging_runner.snapshot()
    for k in ("enabled", "exec_host_side", "failed", "deferred", "permanent_env", "preflight"):
        assert k in snap, f"snapshot missing key {k}"
    # The three non-success classes are distinct keys (code-fail vs transient vs env).
    assert {"failed", "deferred", "permanent_env"} <= set(snap)


def test_tc12_snapshot_never_raises(monkeypatch):
    class Boom:
        def __getattr__(self, name):
            raise RuntimeError("boom")
    monkeypatch.setattr(staging_runner, "settings", Boom())
    assert staging_runner.snapshot()["enabled"] is False


# ---------------------------------------------------------------------------
# R-2 — a held deploy-staging task is NOT rolled back to development by the
# reconciler (the defect would re-appear if it were). AC-3 / AC-11.
# ---------------------------------------------------------------------------
def test_r2_held_deploy_staging_not_rolled_back(monkeypatch):
    tid = _make_task("deploy-staging")
    # No 15-staging-log.md was written (infra-HOLD) -> check_staging_status is red.
    advance = MagicMock()
    monkeypatch.setattr(stage_engine, "advance_stage", advance)

    result = stage_engine.advance_if_gate_passed(
        tid, "deploy-staging", "orchestrator", "ORCH-123", "feature/ORCH-123-x"
    )

    assert result is None                               # red gate -> stay, no advance call
    advance.assert_not_called()                          # NEVER rolled back to development
    conn = get_db()
    stage = conn.execute("SELECT stage FROM tasks WHERE id=?", (tid,)).fetchone()[0]
    conn.close()
    assert stage == "deploy-staging"                     # held in place


# ---------------------------------------------------------------------------
# TC-13 — documentation of the execution boundary (AC-12)
# ---------------------------------------------------------------------------
def _repo_root():
    return os.path.dirname(os.path.dirname(os.path.abspath(__file__)))


def test_tc13_docs_describe_execution_boundary():
    root = _repo_root()
    with open(os.path.join(root, "docs/operations/INFRA.md"), encoding="utf-8") as f:
        infra = f.read().lower()
    with open(os.path.join(root, "docs/architecture/README.md"), encoding="utf-8") as f:
        readme = f.read().lower()
    # INFRA.md fixes the boundary: no docker CLI in the container; docker ops host-side.
    assert "host-side" in infra
    assert "docker" in infra and "ssh" in infra
    assert "orch-123" in infra
    # README describes the host-side staging-runner strategy.
    assert "orch-123" in readme
    assert "host-side" in readme


# ---------------------------------------------------------------------------
# TC-14 — anti-drift ORCH-115: invariants intact (kill-switch -> LLM, shared map)
# ---------------------------------------------------------------------------
def test_tc14_orch115_invariants_intact(monkeypatch):
    # Kill-switch off -> the single LLM transport path is restored (no intercept).
    monkeypatch.setattr(cfg.settings, "staging_runner_enabled", False)
    tid = _make_task("deploy-staging")
    assert staging_runner.should_intercept(_job_dict(1, "deployer", "orchestrator", tid)) is False
    # The exit-code mapping is still the SAME shared self_deploy contract (no drift).
    from src import self_deploy
    for v in (0, 1, None, "x"):
        assert staging_runner.map_exit_code_to_status(v) == self_deploy.map_exit_code_to_status(v)
