"""Contracts for the closed-beta safety net. Fixtures are fictional."""

from __future__ import annotations

import json
import os
import shutil
import subprocess
import sys
import threading
import time
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
WORKFLOW = ROOT / ".gitea" / "workflows" / "backup-production-daily.yml"
RECOVERY_WORKFLOW = ROOT / ".gitea" / "workflows" / "create-production-recovery.yml"
RECOVERY_SCRIPT = ROOT / "deploy" / "run-production-recovery.sh"
PRUNE_SCRIPT = ROOT / "deploy" / "prune-production-daily-backups.sh"
HEALTH_SCRIPT = ROOT / "deploy" / "monitor" / "healthcheck-alert.sh"
DIGEST_SCRIPT = ROOT / "deploy" / "monitor" / "error-digest.sh"
FIND_SCRIPT = ROOT / "deploy" / "find-request-log.sh"
NOTIFY_SCRIPT = ROOT / "deploy" / "monitor" / "notify-feishu.sh"
SERVER_COMPOSE = ROOT / "deploy" / "docker-compose.server.yml"
POSTGRES_COMPOSE = ROOT / "deploy" / "docker-compose.postgres.yml"


def resolve_bash() -> str:
    found = shutil.which("bash")
    git_bash = Path(r"D:\Program Files\Git\bin\bash.exe")
    if os.name == "nt" and git_bash.exists():
        lowered = (found or "").lower()
        if not found or "system32" in lowered or "windowsapps" in lowered:
            return str(git_bash)
    if found:
        return found
    if git_bash.exists():
        return str(git_bash)
    return "bash"


BASH = resolve_bash()


def bash_path(path: Path) -> str:
    if os.name != "nt":
        return str(path)
    completed = subprocess.run(
        [BASH, "-lc", 'cygpath -u "$1"', "cygpath", str(path)],
        check=True,
        capture_output=True,
        text=True,
        encoding="utf-8",
    )
    return completed.stdout.strip()


def run_bash(script: Path, extra: dict[str, str], args: list[str] | None = None) -> subprocess.CompletedProcess[str]:
    env = os.environ.copy()
    env.pop("FEISHU_WEBHOOK_URL", None)
    env.pop("FEISHU_SIGN_SECRET", None)
    env.update(extra)
    env["JYOTISHA_PYTHON"] = bash_path(Path(sys.executable))
    return subprocess.run(
        [BASH, bash_path(script), *(args or [])],
        cwd=ROOT,
        env=env,
        stdin=subprocess.DEVNULL,
        capture_output=True,
        text=True,
        encoding="utf-8",
        errors="replace",
    )


def write_text(path: Path, text: str) -> None:
    path.write_text(text, encoding="utf-8", newline="\n")


def write_backup_record(directory: Path, verified_at: str) -> Path:
    record = directory / "daily-backup-latest.env"
    write_text(
        record,
        "BACKUP_BASENAME=production-daily-20261015T010000Z.dump.enc\n"
        f"BACKUP_SHA256={'a' * 64}\n"
        f"VERIFIED_AT={verified_at}\n"
        "BACKUP_FORMAT=2\n",
    )
    return record


def run_disk_mode(bin_dir: Path, script_env: dict[str, str]) -> subprocess.CompletedProcess[str]:
    return subprocess.run(
        [BASH, "-lc", 'export PATH="$1:$PATH"; exec bash "$2" --mode disk', "bash", bash_path(bin_dir), bash_path(HEALTH_SCRIPT)],
        cwd=ROOT,
        env={**os.environ, **script_env, "JYOTISHA_PYTHON": bash_path(Path(sys.executable))},
        capture_output=True,
        text=True,
        encoding="utf-8",
        errors="replace",
    )


class HealthHandler(BaseHTTPRequestHandler):
    def do_GET(self) -> None:
        if self.path != "/health":
            self.send_response(404)
            self.end_headers()
            return
        kind = getattr(self.server, "kind", "ok")
        if kind == "timeout":
            time.sleep(3)
        commit = getattr(self.server, "commit", "a" * 40)
        if kind == "500":
            status, body = 500, b'{"status":"ok"}'
        elif kind == "missing":
            status, body = 200, b"{}"
        elif kind == "degraded":
            status, body = 200, json.dumps({"status": "degraded", "deployment": {"gitCommit": commit}}).encode()
        else:
            status, body = 200, json.dumps({"status": "ok", "deployment": {"gitCommit": commit}}).encode()
        self.send_response(status)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def log_message(self, fmt: str, *args: object) -> None:
        return


class HookHandler(BaseHTTPRequestHandler):
    def do_POST(self) -> None:
        length = int(self.headers.get("Content-Length", "0"))
        body = self.rfile.read(length)
        self.server.bodies.append(body)
        self.close_connection = True
        code = int(getattr(self.server, "feishu_code", 0))
        reply = json.dumps({"code": code}).encode("utf-8")
        self.send_response(200)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(reply)))
        self.send_header("Connection", "close")
        self.end_headers()
        self.wfile.write(reply)

    def log_message(self, fmt: str, *args: object) -> None:
        return


def serve(handler: type[BaseHTTPRequestHandler]) -> tuple[ThreadingHTTPServer, threading.Thread]:
    server = ThreadingHTTPServer(("127.0.0.1", 0), handler)
    thread = threading.Thread(target=server.serve_forever, daemon=True)
    thread.start()
    return server, thread


def alert_lines(state: Path) -> list[str]:
    log = state / "alerts.log"
    if not log.exists():
        return []
    return [line for line in log.read_text(encoding="utf-8").splitlines() if line]


def test_daily_workflow_is_scheduled_from_main_and_reuses_recovery() -> None:
    text = WORKFLOW.read_text(encoding="utf-8")
    assert "\tif:" not in text
    assert "if: gitea.ref == 'refs/heads/main'" in text
    assert 'cron: "0 20 * * *"' in text
    assert "workflow_dispatch:" in text
    assert 'git show "$GITEA_SHA:deploy/run-production-recovery.sh"' in text
    assert "RECOVERY_BASENAME_PREFIX='production-daily'" in text
    assert "prune-production-daily-backups.sh" in text
    prune_call = text.index("bash '$incoming/prune-production-daily-backups.sh'")
    recovery_call = text.index("bash '$incoming/run-production-recovery.sh'")
    assert prune_call < recovery_call
    assert "production-daily-" in text
    assert 'artifact_retention="${ARTIFACT_RETENTION_DAYS:-14}"' in text
    assert 'local_retention="${LOCAL_RETENTION_DAYS:-7}"' in text
    assert 'process.env["INPUT_RETENTION-DAYS"]=process.env.ARTIFACT_RETENTION_DAYS' in text
    assert "set +x" in text
    assert "pg_dump" not in text
    assert "CHECK_RELEASE_GATE" not in text
    assert "observed_staging_sha" not in text
    assert "STAGING_URL" not in text
    assert "echo \"$SSH_PRIVATE_KEY" not in text
    assert "PRODUCTION_BACKUP_ENCRYPTION_KEY" not in text
    recovery = RECOVERY_WORKFLOW.read_text(encoding="utf-8")
    assert "observed_staging_sha" in recovery
    assert 'process.env["INPUT_RETENTION-DAYS"]="30"' in recovery


def test_recovery_script_keeps_manual_names_unless_daily_prefix() -> None:
    text = RECOVERY_SCRIPT.read_text(encoding="utf-8")
    assert 'backup_prefix="production-pre-migration"' in text
    assert 'production-daily) backup_prefix="production-daily"' in text
    assert "openssl enc -aes-256-cbc -salt -pbkdf2" in text
    assert "usage_percent < 70" in text
    assert "flock -n 9" in text
    assert '[ "$DEPLOY_PATH" = "/opt/jyotisha-production" ]' in text
    assert 'echo "baseline_updated=false" >&2' in text
    assert 'echo "baseline_updated=true" >&2' in text
    assert "daily-backup-latest.env" in text


def install_fake_recovery_bin(bin_dir: Path) -> None:
    bin_dir.mkdir()
    write_text(
        bin_dir / "sudo",
        "#!/usr/bin/env bash\nset -euo pipefail\nif [[ \"${1:-}\" == \"-n\" ]]; then shift; fi\nexec \"$@\"\n",
    )
    write_text(
        bin_dir / "df",
        "#!/usr/bin/env bash\nprintf '%s\\n' 'Filesystem 1024-blocks Used Available Capacity Mounted' 'tmpfs 100 10 90 10% /'\n",
    )
    # Windows cannot chmod pytest's temp directories. These stand-ins only
    # exist on the test PATH. The production script still calls the real commands.
    write_text(
        bin_dir / "chmod",
        "#!/usr/bin/env bash\nexit 0\n",
    )
    # Git Bash on this Windows machine has no flock. The production script still calls it.
    write_text(
        bin_dir / "flock",
        "#!/usr/bin/env bash\nexit 0\n",
    )
    write_text(
        bin_dir / "install",
        """#!/usr/bin/env bash
set -euo pipefail
dirs=()
while [[ $# -gt 0 ]]; do
  case "$1" in
    -d) shift ;;
    -m) shift; shift ;;
    --) shift; break ;;
    -*) shift ;;
    *) dirs+=("$1"); shift ;;
  esac
done
mkdir -p "${dirs[@]}"
""",
    )
    write_text(
        bin_dir / "docker",
        """#!/usr/bin/env bash
set -euo pipefail
cmd="${1:-}"
shift || true
case "$cmd" in
  ps)
    printf '%s\\n' fixturecontainer
    ;;
  inspect)
    printf '%s\\n' sha256:0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef
    ;;
  run)
    exit 0
    ;;
  exec)
    tool=""
    sql=""
    prev=""
    for arg in "$@"; do
      if [[ "$prev" == "-c" ]]; then
        sql="$arg"
      fi
      case "$arg" in
        pg_dump|pg_restore|psql) tool="$arg" ;;
      esac
      prev="$arg"
    done
    case "$tool" in
      pg_dump)
        printf '%s\\n' fixture-dump
        ;;
      pg_restore)
        cat >/dev/null
        ;;
      psql)
        case "$sql" in
          *role_table_grants*) printf '%s\\n' 'reader|SELECT|profiles' ;;
          *column_privileges*) printf '%s\\n' 'reader|SELECT|profiles|id' ;;
          *pg_get_userbyid*) printf '%s\\n' 'profiles|app' ;;
          *identity.users*) printf '%s\\n' '1|2|3|4' ;;
          *"CREATE DATABASE"*|*"DROP DATABASE"*) ;;
          *)
            printf 'unexpected psql\\n' >&2
            exit 1
            ;;
        esac
        ;;
      *)
        printf 'unexpected docker exec\\n' >&2
        exit 1
        ;;
    esac
    ;;
  *)
    printf 'unexpected docker %s\\n' "$cmd" >&2
    exit 1
    ;;
esac
""",
    )
    subprocess.run(
        [BASH, "-lc", 'chmod +x "$1/sudo" "$1/df" "$1/docker" "$1/chmod" "$1/install" "$1/flock"', "chmod", bash_path(bin_dir)],
        check=True,
    )


def write_recovery_env(deploy: Path) -> None:
    write_text(
        deploy / ".env.production.database",
        "\n".join(
            [
                "POSTGRES_DB=fixturedatabase",
                "POSTGRES_USER=fixtureuser",
                "POSTGRES_PASSWORD=fixturepassword",
                "PRODUCTION_BACKUP_ENCRYPTION_KEY=fixtureencryptionkey",
                "",
            ]
        ),
    )


def run_recovery(deploy: Path, bin_dir: Path, prefix: str | None) -> subprocess.CompletedProcess[str]:
    env = os.environ.copy()
    env.pop("FEISHU_WEBHOOK_URL", None)
    env.pop("FEISHU_SIGN_SECRET", None)
    env.pop("RECOVERY_BASENAME_PREFIX", None)
    env["DEPLOY_PATH"] = bash_path(deploy)
    env["RECOVERY_RUN_ID"] = "4242"
    env["JYOTISHA_RECOVERY_FIXTURE"] = "1"
    if prefix:
        env["RECOVERY_BASENAME_PREFIX"] = prefix
    return subprocess.run(
        [
            BASH,
            "-lc",
            'export PATH="$1:$PATH"; exec bash "$2"',
            "bash",
            bash_path(bin_dir),
            bash_path(RECOVERY_SCRIPT),
        ],
        cwd=ROOT,
        env=env,
        stdin=subprocess.DEVNULL,
        capture_output=True,
        text=True,
        encoding="utf-8",
        errors="replace",
    )


def parse_env(text: str) -> dict[str, str]:
    parsed: dict[str, str] = {}
    for line in text.splitlines():
        key, value = line.split("=", 1)
        parsed[key] = value
    return parsed


def test_daily_backup_does_not_replace_the_recovery_point(tmp_path: Path) -> None:
    deploy = tmp_path / "deploy"
    state = deploy / ".state"
    state.mkdir(parents=True)
    write_recovery_env(deploy)
    baseline = state / "recovery-baseline.env"
    original = (
        "RECOVERY_REFERENCE=gitea-actions-run-1/production-pre-migration-20261014T120000Z.dump.enc\n"
        "RECOVERY_CREATED_AT=2026-10-14T12:00:00Z\n"
        "RECOVERY_SHA256=" + ("ab" * 32) + "\n"
        "RESTORE_VERIFIED=true\n"
        "RECOVERY_FORMAT=2\n"
        "VERIFY_FILE=production-pre-migration-20261014T120000Z-restore-verify.json\n"
    )
    write_text(baseline, original)
    subprocess.run(
        [BASH, "-lc", 'touch -d "@$1" "$2"', "touch", "1700000000", bash_path(baseline)],
        check=True,
    )
    before_ns = baseline.stat().st_mtime_ns
    before_bytes = baseline.read_bytes()
    bin_dir = tmp_path / "bin"
    install_fake_recovery_bin(bin_dir)
    completed = run_recovery(deploy, bin_dir, "production-daily")
    assert completed.returncode == 0, completed.stderr
    assert baseline.read_bytes() == before_bytes
    assert baseline.stat().st_mtime_ns == before_ns
    assert "baseline_updated" not in completed.stdout
    assert "baseline_updated=false" in completed.stderr.splitlines()
    assert "baseline_updated=true" not in completed.stderr
    assert len(completed.stdout.splitlines()) == 7
    stdout_env = parse_env(completed.stdout)
    assert stdout_env["BACKUP_BASENAME"].startswith("production-daily-")
    daily = state / "daily-backup-latest.env"
    daily_env = parse_env(daily.read_text(encoding="utf-8"))
    assert set(daily_env) == {"BACKUP_BASENAME", "BACKUP_SHA256", "VERIFIED_AT", "BACKUP_FORMAT"}
    assert daily_env["BACKUP_FORMAT"] == "2"
    assert daily_env["BACKUP_BASENAME"] == stdout_env["BACKUP_BASENAME"]
    assert daily_env["BACKUP_SHA256"] == stdout_env["RECOVERY_SHA256"]
    daily_text = daily.read_text(encoding="utf-8")
    assert "fixturepassword" not in daily_text
    assert "fixtureencryptionkey" not in daily_text
    assert "fixturedatabase" not in daily_text
    verify = deploy / "backups" / stdout_env["VERIFY_BASENAME"]
    payload = json.loads(verify.read_text(encoding="utf-8"))
    assert payload["checks"]["table_grants_match"] is True
    assert payload["checks"]["column_privileges_match"] is True
    assert payload["checks"]["table_owners_match"] is True
    assert "fixturepassword" not in verify.read_text(encoding="utf-8")


def test_manual_recovery_still_writes_format_two(tmp_path: Path) -> None:
    deploy = tmp_path / "deploy"
    state = deploy / ".state"
    state.mkdir(parents=True)
    write_recovery_env(deploy)
    sentinel = state / "daily-backup-latest.env"
    write_text(sentinel, "BACKUP_BASENAME=keep\nBACKUP_FORMAT=2\n")
    before = sentinel.read_bytes()
    bin_dir = tmp_path / "bin"
    install_fake_recovery_bin(bin_dir)
    completed = run_recovery(deploy, bin_dir, None)
    assert completed.returncode == 0, completed.stderr
    assert "baseline_updated=true" in completed.stderr.splitlines()
    assert "baseline_updated=false" not in completed.stderr
    assert len(completed.stdout.splitlines()) == 7
    baseline_text = (state / "recovery-baseline.env").read_text(encoding="utf-8")
    baseline = parse_env(baseline_text)
    assert baseline["RECOVERY_FORMAT"] == "2"
    assert baseline["VERIFY_FILE"].startswith("production-pre-migration-")
    assert baseline["RECOVERY_REFERENCE"].startswith("gitea-actions-run-4242/production-pre-migration-")
    assert "fixturepassword" not in baseline_text
    assert "fixtureencryptionkey" not in baseline_text
    assert parse_env(completed.stdout)["BACKUP_BASENAME"].startswith("production-pre-migration-")
    assert sentinel.read_bytes() == before


def test_release_migration_and_rollback_ignore_daily_backup_record() -> None:
    readers = [
        ROOT / ".gitea" / "workflows" / "deploy-production.yml",
        ROOT / ".gitea" / "workflows" / "migrate-production-database.yml",
        ROOT / "deploy" / "run-production-deploy.sh",
        ROOT / "deploy" / "run-production-migration.sh",
    ]
    for path in readers:
        text = path.read_text(encoding="utf-8")
        assert "daily-backup-latest.env" not in text, path.name
        assert "recovery-baseline.env" in text
    for path in (WORKFLOW, RECOVERY_WORKFLOW):
        text = path.read_text(encoding="utf-8")
        assert "JYOTISHA_RECOVERY_FIXTURE" not in text
        assert '[[ "$(wc -l <"$ssh_root/recovery-output.env" | tr -d \' \')" == 7 ]]' in text


def test_prune_deletes_only_expired_daily_files(tmp_path: Path) -> None:
    backup = tmp_path / "backups"
    nested = backup / "nested"
    nested.mkdir(parents=True)
    now = 1_700_000_000
    limit = 7 * 86400
    names = {
        "production-daily-20261001T000000Z.dump.enc": now - limit - 60,
        "production-daily-20261001T000000Z-restore-verify.json": now - limit - 60,
        "production-daily-20261007T000000Z.dump.enc": now - limit,
        "production-daily-20261007T000000Z-restore-verify.json": now - limit,
        "production-daily-20261014T000000Z.dump.enc": now - 3600,
        "production-daily-20261014T000000Z-restore-verify.json": now - 3600,
        "production-pre-migration-20261001T000000Z.dump.enc": now - limit - 3600,
        "production-pre-migration-20261001T000000Z-restore-verify.json": now - limit - 3600,
        "production-daily-notes.txt": now - limit - 3600,
        "production-daily-20261001T000000Z.dump.enc.bak": now - limit - 3600,
        "unrelated.txt": now - limit - 3600,
    }
    for name, epoch in names.items():
        path = backup / name
        write_text(path, "fictional\n")
        subprocess.run(
            [BASH, "-lc", 'touch -d "@$1" "$2"', "touch", str(epoch), bash_path(path)],
            check=True,
        )
    write_text(nested / "production-daily-20261001T000000Z.dump.enc", "nested\n")
    completed = run_bash(PRUNE_SCRIPT, {}, [bash_path(backup), "7", str(now)])
    assert completed.returncode == 0, completed.stderr
    deleted = completed.stdout.splitlines()
    assert deleted == ["production-daily-20261001T000000Z.dump.enc"]
    assert not (backup / "production-daily-20261001T000000Z.dump.enc").exists()
    assert not (backup / "production-daily-20261001T000000Z-restore-verify.json").exists()
    for name in names:
        if name.startswith("production-daily-20261001T000000Z.dump") or name.startswith(
            "production-daily-20261001T000000Z-restore"
        ):
            continue
        assert (backup / name).exists(), name
    assert (nested / "production-daily-20261001T000000Z.dump.enc").exists()


def test_health_alerts_after_three_failures_and_recovery_once(tmp_path: Path) -> None:
    server, _thread = serve(HealthHandler)
    server.kind = "500"
    state = tmp_path / "state"
    env = {
        "JYOTISHA_MONITOR_STATE_DIR": bash_path(state),
        "JYOTISHA_MONITOR_ENV": bash_path(tmp_path / "missing.env"),
        "JYOTISHA_HEALTH_URL": f"http://127.0.0.1:{server.server_address[1]}/health",
        "JYOTISHA_HEALTH_TIMEOUT": "2",
    }
    try:
        for _ in range(2):
            completed = run_bash(HEALTH_SCRIPT, env, ["--mode", "health"])
            assert completed.returncode == 0, completed.stderr
        assert alert_lines(state) == []
        completed = run_bash(HEALTH_SCRIPT, env, ["--mode", "health"])
        assert completed.returncode == 0, completed.stderr
        lines = alert_lines(state)
        assert len(lines) == 1
        assert "errorCode=health_down" in lines[0]
        assert "count=3" in lines[0]
        assert "route=/api/health" in lines[0]
        server.kind = "ok"
        server.commit = "b" * 40
        completed = run_bash(HEALTH_SCRIPT, env, ["--mode", "health"])
        assert completed.returncode == 0, completed.stderr
        lines = alert_lines(state)
        assert len(lines) == 2
        assert "errorCode=recovered" in lines[1]
        completed = run_bash(HEALTH_SCRIPT, env, ["--mode", "health"])
        assert completed.returncode == 0, completed.stderr
        assert len(alert_lines(state)) == 2
    finally:
        server.shutdown()


def test_health_timeout_and_missing_fields_count_as_failures(tmp_path: Path) -> None:
    server, _thread = serve(HealthHandler)
    server.kind = "timeout"
    state = tmp_path / "state"
    env = {
        "JYOTISHA_MONITOR_STATE_DIR": bash_path(state),
        "JYOTISHA_MONITOR_ENV": bash_path(tmp_path / "missing.env"),
        "JYOTISHA_HEALTH_URL": f"http://127.0.0.1:{server.server_address[1]}/health",
        "JYOTISHA_HEALTH_TIMEOUT": "1",
    }
    try:
        completed = run_bash(HEALTH_SCRIPT, env, ["--mode", "health"])
        assert completed.returncode == 0, completed.stderr
        assert alert_lines(state) == []
        server.kind = "missing"
        run_bash(HEALTH_SCRIPT, env, ["--mode", "health"])
        completed = run_bash(HEALTH_SCRIPT, env, ["--mode", "health"])
        assert completed.returncode == 0, completed.stderr
        assert any("errorCode=health_down" in line for line in alert_lines(state))
    finally:
        server.shutdown()


def test_commit_change_alerts_once_outside_release_window(tmp_path: Path) -> None:
    server, _thread = serve(HealthHandler)
    server.kind = "ok"
    server.commit = "a" * 40
    state = tmp_path / "state"
    env = {
        "JYOTISHA_MONITOR_STATE_DIR": bash_path(state),
        "JYOTISHA_MONITOR_ENV": bash_path(tmp_path / "missing.env"),
        "JYOTISHA_HEALTH_URL": f"http://127.0.0.1:{server.server_address[1]}/health",
        "JYOTISHA_HEALTH_TIMEOUT": "2",
    }
    try:
        assert run_bash(HEALTH_SCRIPT, env, ["--mode", "health"]).returncode == 0
        assert alert_lines(state) == []
        server.commit = "b" * 40
        assert run_bash(HEALTH_SCRIPT, env, ["--mode", "health"]).returncode == 0
        lines = alert_lines(state)
        assert len(lines) == 1
        assert "errorCode=commit_changed" in lines[0]
        assert "b" * 40 not in lines[0]
        assert run_bash(HEALTH_SCRIPT, env, ["--mode", "health"]).returncode == 0
        assert len(alert_lines(state)) == 1
        marker = state / "release-window"
        state.mkdir(parents=True, exist_ok=True)
        write_text(marker, "open\n")
        server.commit = "c" * 40
        assert run_bash(HEALTH_SCRIPT, env, ["--mode", "health"]).returncode == 0
        assert len(alert_lines(state)) == 1
    finally:
        server.shutdown()


def test_disk_alerts_once_and_no_webhook_stays_local(tmp_path: Path) -> None:
    bin_dir = tmp_path / "bin"
    bin_dir.mkdir()
    write_text(
        bin_dir / "df",
        "#!/usr/bin/env bash\nprintf '%s\\n' 'Filesystem 1024-blocks Used Available Capacity Mounted' \"overlay 100 90 10 ${FAKE_DISK_PERCENT:?}% /\"\n",
    )
    subprocess.run([BASH, "-lc", 'chmod +x "$1/df"', "chmod", bash_path(bin_dir)], check=True)
    state = tmp_path / "state"
    record = write_backup_record(tmp_path, "2026-10-15T01:00:00Z")
    env = {
        "JYOTISHA_MONITOR_STATE_DIR": bash_path(state),
        "JYOTISHA_MONITOR_ENV": bash_path(tmp_path / "missing.env"),
        "JYOTISHA_DAILY_BACKUP_RECORD": bash_path(record),
        "JYOTISHA_NOW": "2026-10-15T01:00:00Z",
        "FAKE_DISK_PERCENT": "81",
    }
    env.pop("FEISHU_WEBHOOK_URL", None)
    env.pop("FEISHU_SIGN_SECRET", None)
    completed = run_disk_mode(bin_dir, env)
    assert completed.returncode == 0, completed.stderr
    assert "errorCode=disk_high" in alert_lines(state)[0]
    assert "count=81" in alert_lines(state)[0]
    env["FAKE_DISK_PERCENT"] = "81"
    again = run_disk_mode(bin_dir, env)
    assert again.returncode == 0, again.stderr
    assert len(alert_lines(state)) == 1
    env["FAKE_DISK_PERCENT"] = "10"
    recovered = run_disk_mode(bin_dir, env)
    assert recovered.returncode == 0, recovered.stderr
    assert len(alert_lines(state)) == 2
    assert "errorCode=recovered" in alert_lines(state)[1]


def test_feishu_payload_keeps_only_allowed_fields(tmp_path: Path) -> None:
    hook, _thread = serve(HookHandler)
    hook.bodies = []
    health, _thread = serve(HealthHandler)
    health.kind = "500"
    state = tmp_path / "state"
    secret = "fixture-sign-secret"
    env_file = tmp_path / "monitor.env"
    write_text(env_file, f"FEISHU_WEBHOOK_URL=http://127.0.0.1:{hook.server_address[1]}/hook\nFEISHU_SIGN_SECRET={secret}\n")
    os.chmod(env_file, 0o600)
    if os.name != "nt":
        assert (env_file.stat().st_mode & 0o777) == 0o600
    notify = NOTIFY_SCRIPT.read_text(encoding="utf-8")
    assert '[ "$mode" = "600" ]' in notify
    assert "curl --silent --show-error --fail" in notify
    env = {
        "JYOTISHA_MONITOR_STATE_DIR": bash_path(state),
        "JYOTISHA_MONITOR_ENV": bash_path(env_file),
        "JYOTISHA_HEALTH_URL": f"http://127.0.0.1:{health.server_address[1]}/health",
        "JYOTISHA_HEALTH_TIMEOUT": "2",
    }
    try:
        for _ in range(3):
            completed = run_bash(HEALTH_SCRIPT, env, ["--mode", "health"])
            assert completed.returncode == 0, completed.stderr
        assert len(hook.bodies) == 1
        payload = json.loads(hook.bodies[0].decode("utf-8"))
        assert set(payload) == {"timestamp", "sign", "msg_type", "content"}
        assert payload["msg_type"] == "text"
        text = payload["content"]["text"]
        assert secret not in text
        assert secret not in hook.bodies[0].decode("utf-8")
        assert "errorCode=health_down" in text
        assert "@" not in text
        expected = subprocess.run(
            [
                sys.executable,
                "-c",
                "import base64,hashlib,hmac,os,sys; ts=sys.argv[1]; secret=sys.argv[2]; key=f'{ts}\\n{secret}'.encode(); print(base64.b64encode(hmac.new(key, digestmod=hashlib.sha256).digest()).decode())",
                payload["timestamp"],
                secret,
            ],
            check=True,
            capture_output=True,
            text=True,
        ).stdout.strip()
        assert payload["sign"] == expected
    finally:
        hook.shutdown()
        health.shutdown()


def test_feishu_rejection_is_not_recorded_as_sent(tmp_path: Path) -> None:
    hook, _thread = serve(HookHandler)
    hook.bodies = []
    hook.feishu_code = 19001
    health, _thread = serve(HealthHandler)
    health.kind = "500"
    state = tmp_path / "state"
    env_file = tmp_path / "monitor.env"
    write_text(env_file, f"FEISHU_WEBHOOK_URL=http://127.0.0.1:{hook.server_address[1]}/hook\n")
    os.chmod(env_file, 0o600)
    env = {
        "JYOTISHA_MONITOR_STATE_DIR": bash_path(state),
        "JYOTISHA_MONITOR_ENV": bash_path(env_file),
        "JYOTISHA_HEALTH_URL": f"http://127.0.0.1:{health.server_address[1]}/health",
        "JYOTISHA_HEALTH_TIMEOUT": "2",
    }
    try:
        assert run_bash(HEALTH_SCRIPT, env, ["--mode", "health"]).returncode == 0
        assert run_bash(HEALTH_SCRIPT, env, ["--mode", "health"]).returncode == 0
        rejected = run_bash(HEALTH_SCRIPT, env, ["--mode", "health"])
        assert rejected.returncode != 0
        assert len(hook.bodies) == 1
        assert alert_lines(state) == []
        assert "health_down" not in rejected.stdout
    finally:
        hook.shutdown()
        health.shutdown()


def uuid(index: int) -> str:
    return f"00000000-0000-4000-8000-{index:012d}"


def test_error_digest_drops_personal_text_and_respects_thresholds(tmp_path: Path) -> None:
    safe = [
        f"2026-10-15T00:00:{index:02d}Z web [agent-observability] {json.dumps({'requestId': uuid(index), 'errorCode': 'settlement_failed'})}"
        for index in range(1, 11)
    ]
    unsafe = [
        "2026-10-15T00:01:00Z web person@example.com errorCode=should_drop",
        "2026-10-15T00:01:01Z web note 1900-01-01 birth_date",
        "2026-10-15T00:01:02Z web 对话内容 今天工作不顺",
    ]
    fixture = tmp_path / "logs.txt"
    write_text(fixture, "\n".join([*safe, *unsafe]) + "\n")
    state = tmp_path / "state"
    env = {
        "JYOTISHA_MONITOR_STATE_DIR": bash_path(state),
        "JYOTISHA_MONITOR_ENV": bash_path(tmp_path / "missing.env"),
        "JYOTISHA_LOG_FIXTURE": bash_path(fixture),
        "JYOTISHA_NOW": "2026-10-15T01:00:00Z",
    }
    below = tmp_path / "below.txt"
    write_text(below, "\n".join(safe[:9]) + "\n")
    quiet = run_bash(DIGEST_SCRIPT, {**env, "JYOTISHA_LOG_FIXTURE": bash_path(below)}, ["--window"])
    assert quiet.returncode == 0, quiet.stderr
    assert alert_lines(state) == []
    completed = run_bash(DIGEST_SCRIPT, env, ["--window"])
    assert completed.returncode == 0, completed.stderr
    lines = alert_lines(state)
    assert len(lines) == 1
    assert "errorCode=settlement_failed" in lines[0]
    assert "count=10" in lines[0]
    assert lines[0].count(",") == 4
    assert uuid(6) not in lines[0]
    assert "person@example.com" not in lines[0]
    assert "1900-01-01" not in lines[0]
    assert "今天工作不顺" not in lines[0]
    assert "should_drop" not in lines[0]
    soon = run_bash(DIGEST_SCRIPT, {**env, "JYOTISHA_NOW": "2026-10-15T01:29:59Z"}, ["--window"])
    assert soon.returncode == 0, soon.stderr
    assert len(alert_lines(state)) == 1
    later = run_bash(DIGEST_SCRIPT, {**env, "JYOTISHA_NOW": "2026-10-15T01:30:00Z"}, ["--window"])
    assert later.returncode == 0, later.stderr
    assert len(alert_lines(state)) == 2
    assert "errorCode=settlement_failed" in alert_lines(state)[1]

    fivexx = tmp_path / "fivexx.txt"
    write_text(fivexx, "\n".join(f"2026-10-15T00:02:{index:02d}Z api status=502" for index in range(20)) + "\n")
    state_5xx = tmp_path / "state5"
    completed = run_bash(
        DIGEST_SCRIPT,
        {**env, "JYOTISHA_MONITOR_STATE_DIR": bash_path(state_5xx), "JYOTISHA_LOG_FIXTURE": bash_path(fivexx)},
        ["--window"],
    )
    assert completed.returncode == 0, completed.stderr
    joined = "\n".join(alert_lines(state_5xx))
    assert "errorCode=http_5xx" in joined
    assert "count=20" in joined


def test_daily_digest_summarizes_previous_shanghai_day(tmp_path: Path) -> None:
    fixture = tmp_path / "daily.txt"
    write_text(
        fixture,
        "\n".join(
            [
                '2026-10-14T02:00:00Z api [agent-observability] {"requestId":"00000000-0000-4000-8000-000000000001","errorCode":"settlement_failed"}',
                '2026-10-15T01:30:00Z api [agent-observability] {"requestId":"00000000-0000-4000-8000-000000000002","errorCode":"settlement_failed"}',
            ]
        )
        + "\n",
    )
    state = tmp_path / "state"
    completed = run_bash(
        DIGEST_SCRIPT,
        {
            "JYOTISHA_MONITOR_STATE_DIR": bash_path(state),
            "JYOTISHA_MONITOR_ENV": bash_path(tmp_path / "missing.env"),
            "JYOTISHA_LOG_FIXTURE": bash_path(fixture),
            "JYOTISHA_NOW": "2026-10-15T01:00:00Z",
        },
        ["--daily"],
    )
    assert completed.returncode == 0, completed.stderr
    lines = alert_lines(state)
    assert len(lines) == 1
    assert "count=1" in lines[0]
    assert "00000000-0000-4000-8000-000000000001" in lines[0]
    assert "00000000-0000-4000-8000-000000000002" not in lines[0]


def test_find_request_log_sorts_and_rejects_injection(tmp_path: Path) -> None:
    log_dir = tmp_path / "logs"
    log_dir.mkdir()
    request_id = "11111111-2222-4333-8444-555555555555"
    write_text(log_dir / "caddy.log", f"2026-10-15T03:00:00Z caddy {request_id}\n")
    write_text(log_dir / "web.log", f"2026-10-15T01:00:00Z web {request_id}\nignore\n")
    write_text(log_dir / "api.log", f"2026-10-15T02:00:00Z api {request_id}\n")
    canary = tmp_path / "canary.txt"
    write_text(canary, "keep\n")
    found = run_bash(FIND_SCRIPT, {"JYOTISHA_LOG_DIR": bash_path(log_dir)}, [request_id, "--since", "24h"])
    assert found.returncode == 0, found.stderr
    assert found.stdout.splitlines() == [
        f"2026-10-15T01:00:00Z web {request_id}",
        f"2026-10-15T02:00:00Z api {request_id}",
        f"2026-10-15T03:00:00Z caddy {request_id}",
    ]
    rejected = run_bash(FIND_SCRIPT, {"JYOTISHA_LOG_DIR": bash_path(log_dir)}, ["; rm -rf /", "--since", "24h"])
    assert rejected.returncode == 2
    assert canary.read_text(encoding="utf-8") == "keep\n"
    injected = run_bash(FIND_SCRIPT, {"JYOTISHA_LOG_DIR": bash_path(log_dir)}, ["$(rm canary.txt)", "--since", "1h"])
    assert injected.returncode == 2
    assert canary.exists()
    bad_since = run_bash(FIND_SCRIPT, {"JYOTISHA_LOG_DIR": bash_path(log_dir)}, [request_id, "--since", "24h;rm"])
    assert bad_since.returncode == 2
    compose_log = tmp_path / "compose.log"
    write_text(
        compose_log,
        "\n".join(
            [
                f"web-1  | 2026-10-15T01:00:00Z web {request_id}",
                f"api-1  | 2026-10-15T03:00:00Z api {request_id}",
                f"caddy-1  | 2026-10-15T02:00:00Z caddy {request_id}",
                f"api-1  | ignore {request_id}",
            ]
        )
        + "\n",
    )
    composed = run_bash(
        FIND_SCRIPT,
        {"JYOTISHA_COMPOSE_LOG_FIXTURE": bash_path(compose_log)},
        [request_id, "--since", "24h"],
    )
    assert composed.returncode == 0, composed.stderr
    assert composed.stdout.splitlines() == [
        f"2026-10-15T01:00:00Z web {request_id}",
        f"2026-10-15T02:00:00Z caddy {request_id}",
        f"2026-10-15T03:00:00Z api {request_id}",
    ]


def service_logging_blocks(text: str) -> dict[str, str]:
    blocks: dict[str, str] = {}
    service: str | None = None
    in_services = False
    capturing = False
    buf: list[str] = []
    for line in text.splitlines():
        if line.startswith("services:"):
            in_services = True
            continue
        if in_services and line and not line.startswith(" "):
            in_services = False
        if not in_services:
            continue
        if line.startswith("  ") and not line.startswith("    ") and line.endswith(":"):
            if capturing and service is not None:
                blocks[service] = "\n".join(buf)
            service = line.strip()[:-1]
            capturing = False
            buf = []
            continue
        if service and line == "    logging:":
            capturing = True
            buf = [line]
            continue
        if capturing:
            if line.startswith("    ") and not line.startswith("      "):
                blocks[service] = "\n".join(buf)
                capturing = False
                buf = []
            else:
                buf.append(line)
    if capturing and service is not None:
        blocks[service] = "\n".join(buf)
    return blocks


def test_compose_logging_is_present_and_identical() -> None:
    expected = "\n".join(
        [
            "    logging:",
            "      driver: json-file",
            "      options:",
            '        max-size: "20m"',
            '        max-file: "10"',
        ]
    )
    expected_names = {
        SERVER_COMPOSE: {"api", "web", "caddy"},
        POSTGRES_COMPOSE: {"postgres", "migrator", "migration-checker"},
    }
    for path, names in expected_names.items():
        blocks = service_logging_blocks(path.read_text(encoding="utf-8"))
        assert set(blocks) == names
        for name in names:
            assert blocks[name] == expected, name


def test_monitor_commands_use_production_docker() -> None:
    for script in (DIGEST_SCRIPT, FIND_SCRIPT):
        text = script.read_text(encoding="utf-8")
        assert "sudo -n docker compose -p jyotisha-production" in text
        assert "export DATABASE_ENV_FILE='../.env.production.database'" in text
        assert "export APP_ENV_FILE='../.env.production'" in text
        assert "docker-compose.server.yml" in text
        assert "docker-compose.postgres.yml" in text
        assert "docker-compose.production.yml" in text
        for line in text.splitlines():
            stripped = line.strip()
            if stripped.startswith("docker ") or stripped.startswith("docker compose"):
                raise AssertionError(stripped)
    finder = FIND_SCRIPT.read_text(encoding="utf-8")
    strip_at = finder.index("sed -E 's/^[^|]*\\| //'")
    sort_at = finder.index("sort", strip_at)
    assert strip_at < sort_at


def test_safety_net_is_on_the_quick_gate() -> None:
    text = (ROOT / "scripts" / "run_quality_gate.py").read_text(encoding="utf-8")
    assert '"tests/test_beta_safety_net.py"' in text


def install_fake_df(bin_dir: Path) -> None:
    bin_dir.mkdir()
    write_text(
        bin_dir / "df",
        "#!/usr/bin/env bash\nprintf '%s\\n' 'Filesystem 1024-blocks Used Available Capacity Mounted' \"overlay 100 10 90 ${FAKE_DISK_PERCENT:?}% /\"\n",
    )
    subprocess.run([BASH, "-lc", 'chmod +x "$1/df"', "chmod", bash_path(bin_dir)], check=True)


def test_stale_daily_backup_alerts_once_until_a_fresh_record(tmp_path: Path) -> None:
    bin_dir = tmp_path / "bin"
    install_fake_df(bin_dir)
    state = tmp_path / "state"
    now = "2026-10-15T12:00:00Z"
    env = {
        "JYOTISHA_MONITOR_STATE_DIR": bash_path(state),
        "JYOTISHA_MONITOR_ENV": bash_path(tmp_path / "missing.env"),
        "JYOTISHA_NOW": now,
        "JYOTISHA_DAILY_BACKUP_RECORD": bash_path(tmp_path / "missing-record.env"),
        "FAKE_DISK_PERCENT": "10",
    }
    missing = run_disk_mode(bin_dir, env)
    assert missing.returncode == 0, missing.stderr
    assert alert_lines(state) == [
        line for line in alert_lines(state) if "errorCode=backup_stale" in line and "count=36" in line
    ]
    assert len(alert_lines(state)) == 1
    assert "production-daily" not in alert_lines(state)[0]
    assert ".env" not in alert_lines(state)[0]
    again = run_disk_mode(bin_dir, env)
    assert again.returncode == 0, again.stderr
    assert len(alert_lines(state)) == 1

    exact = tmp_path / "exact"
    exact.mkdir()
    env["JYOTISHA_MONITOR_STATE_DIR"] = bash_path(exact)
    env["JYOTISHA_DAILY_BACKUP_RECORD"] = bash_path(write_backup_record(exact, "2026-10-14T00:00:00Z"))
    boundary = run_disk_mode(bin_dir, env)
    assert boundary.returncode == 0, boundary.stderr
    assert alert_lines(exact) == []

    stale_state = tmp_path / "stale-state"
    old_dir = tmp_path / "old"
    old_dir.mkdir()
    env["JYOTISHA_MONITOR_STATE_DIR"] = bash_path(stale_state)
    env["JYOTISHA_DAILY_BACKUP_RECORD"] = bash_path(write_backup_record(old_dir, "2026-10-13T23:00:00Z"))
    old = run_disk_mode(bin_dir, env)
    assert old.returncode == 0, old.stderr
    assert len(alert_lines(stale_state)) == 1
    assert "errorCode=backup_stale" in alert_lines(stale_state)[0]
    assert "count=37" in alert_lines(stale_state)[0]
    fresh_dir = tmp_path / "fresh"
    fresh_dir.mkdir()
    env["JYOTISHA_DAILY_BACKUP_RECORD"] = bash_path(write_backup_record(fresh_dir, now))
    recovered = run_disk_mode(bin_dir, env)
    assert recovered.returncode == 0, recovered.stderr
    assert len(alert_lines(stale_state)) == 2
    assert "errorCode=recovered" in alert_lines(stale_state)[1]


def test_shell_scripts_pass_bash_n() -> None:
    for script in (PRUNE_SCRIPT, HEALTH_SCRIPT, DIGEST_SCRIPT, FIND_SCRIPT, NOTIFY_SCRIPT, RECOVERY_SCRIPT):
        completed = subprocess.run(
            [BASH, "-n", bash_path(script)],
            capture_output=True,
            text=True,
            encoding="utf-8",
        )
        assert completed.returncode == 0, script.name + completed.stderr