"""Real disposable r71b maintenance lifecycle contracts."""

from __future__ import annotations

import hashlib
import json
import os
from datetime import date, datetime
from pathlib import Path

import pytest
from pydantic import JsonValue, TypeAdapter

from gateway.platforms.nutrition_weekly_maintenance_contract import (
    Topic59MaintenanceScopeV1,
    canonical_json,
)
from scripts import (
    nutricoach_v150_r71b_maintenance_transaction as transaction,
)
from scripts import nutricoach_v150_r71b_sealed_host as sealed_host
from scripts.nutricoach_v150_concrete_host import ConcreteLiveHost
from scripts.nutricoach_v150_phase_journal import PhaseJournal
from scripts.nutricoach_v150_r71b_package import R71bPackageArtifacts, derive_r71b_package
from scripts.nutricoach_v150_sealed_authority import GlobalLedger
from scripts.nutricoach_v150_sealed_controller import execute_disposable, recover_pending
from scripts.nutricoach_v150_sealed_target import DisposableService
from tests.test_nutricoach_v150_sealed_controller import target_fixture

_OBJECT = TypeAdapter(dict[str, JsonValue])


def _artifacts(candidate_digest: str) -> R71bPackageArtifacts:
    scope = Topic59MaintenanceScopeV1(
        candidate_digest=candidate_digest,
        config_digest="2" * 64,
        route_digest="3" * 64,
        customer_identity_digest="4" * 64,
        card_slot="5" * 64,
        kst_day=date(2026, 9, 1),
        not_before=datetime.fromisoformat("2026-09-01T16:00:00+09:00"),
        expires_at=datetime.fromisoformat("2026-09-01T16:10:00+09:00"),
    )
    binding: dict[str, JsonValue] = _OBJECT.validate_python({
        "schema": "nutricoach-v150-r71b-package-binding-inputs-v1",
        "product_generation": "r71",
        "package_namespace": "r71b-maintenance",
        "authority_id": "nutricoach-v150-v15-runtime-authority-r71b-maintenance",
        "paths": {"observer_root": "/tmp/observer-r71"},
        "candidate": {"candidate_digest": candidate_digest},
        "wheels": [],
        "controller": {"files": {}},
        "protected_inventory": {"explicit_maintenance_volatile_paths": []},
        "dependency_snapshot": {"dependency_snapshot_sha256": "6" * 64},
        "authority_baseline": {"candidate_digest": "7" * 64},
        "target_binding_base": {"profile_root": "/tmp/profile"},
        "service_preparation": {"ActiveState": "active"},
        "maintenance_scope": scope.model_dump(mode="json", by_alias=True),
        "maintenance_scope_path": "maintenance/maintenance-scope.json",
        "maintenance_scope_file_sha256": "8" * 64,
        "maintenance_scope_digest": "9" * 64,
        "no_send_oracle": {"schema": "nutricoach-r71b-no-send-oracle-v1"},
        "no_send_oracle_path": "maintenance/no-send-oracle.json",
        "no_send_oracle_file_sha256": "a" * 64,
        "diagnosed_blocker_evidence": [],
    })
    return derive_r71b_package(binding, scope)


def _r70_job(*, status: str, error: str | None) -> dict[str, JsonValue]:
    return {
        "id": transaction.KNOWN_R70_CRON_JOB_ID,
        "name": transaction.KNOWN_R70_CRON_JOB_NAME,
        "schedule_display": transaction.KNOWN_R70_CRON_SCHEDULE,
        "inline_card": "nutrition-coaching-tick",
        "deliver": "local",
        "enabled": True,
        "state": "scheduled",
        "last_status": status,
        "last_error": error,
        "last_run_at": "2026-09-01T16:01:00+09:00",
    }


def _write_audit(path: Path, artifacts: R71bPackageArtifacts) -> None:
    hold = artifacts.hold
    audit: dict[str, JsonValue] = {
        "schema": "nutricoach-topic59-maintenance-skip-audit-v1",
        "hold_id": hold.hold_id,
        "hold_sha256": artifacts.hold_sha256,
        "authority_id": artifacts.authority.authority_id,
        "package_binding_digest": artifacts.package_binding_digest,
        "candidate_digest": hold.candidate_digest,
        "config_digest": hold.config_digest,
        "route_digest": hold.route_digest,
        "customer_identity_digest": hold.customer_identity_digest,
        "card_slot": hold.card_slot,
        "kst_day": hold.kst_day.isoformat(),
        "recorded_at": "2026-09-01T16:01:00+09:00",
        "terminal_state": "maintenance_skipped",
        "disposition": "noop_maintenance",
        "sent": False,
        "delivered": False,
        "message_id": None,
        "provider_receipt": None,
        "provider_calls": 0,
        "network_calls": 0,
    }
    audit["row_digest"] = hashlib.sha256(canonical_json(audit)).hexdigest()
    _ = path.write_bytes(canonical_json(audit) + b"\n")


def test_disposable_maintenance_reaches_committed_with_successor_postimage(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    # Given
    _ = target_fixture(tmp_path)
    service = DisposableService()
    host = ConcreteLiveHost.disposable(tmp_path, service)
    artifacts = _artifacts(host.candidate_digest)
    profile = host.paths.profile
    cron = profile / "cron/jobs.json"
    cron.parent.mkdir()
    initial_error = transaction.KNOWN_R70_CRON_ERROR
    _ = cron.write_text(
        json.dumps({"jobs": [_r70_job(status="error", error=initial_error)]}),
        encoding="utf-8",
    )
    ledger = profile / "data/weekly-operations-topic59.jsonl"
    _ = ledger.write_bytes(b"historical sending rows\n")
    evidence = tmp_path / "task-10-cron-no-send-recovery-analysis.json"
    _ = evidence.write_bytes(b"accepted r70 exception\n")
    r70_candidate = tmp_path / "r70-candidate"
    _ = r70_candidate.write_text("b" * 64 + "\n", encoding="utf-8")
    authority = tmp_path / "maintenance-authority.json"
    _ = authority.write_bytes(artifacts.authority_bytes)
    binding: dict[str, JsonValue] = {
        "maintenance_authority_path": str(authority),
        "maintenance_authority_sha256": artifacts.authority_sha256,
        "package_binding_digest": artifacts.package_binding_digest,
        "candidate_digest": host.candidate_digest,
        "maintenance_ledger_path": str(ledger),
        "maintenance_ledger_sha256": hashlib.sha256(ledger.read_bytes()).hexdigest(),
        "maintenance_output_directory": str(profile / "cron/output" / transaction.KNOWN_R70_CRON_JOB_ID),
        "r70_error_evidence_path": str(evidence),
        "r70_error_evidence_sha256": hashlib.sha256(evidence.read_bytes()).hexdigest(),
        "authority_baseline": {"candidate_path": str(r70_candidate)},
        "package_digest": artifacts.final_package_digest,
    }
    evidence_sha = hashlib.sha256(evidence.read_bytes()).hexdigest()
    for module in (transaction, sealed_host):
        monkeypatch.setattr(module, "KNOWN_R70_CRON_ERROR_EVIDENCE_SHA256", evidence_sha)
        monkeypatch.setattr(module, "KNOWN_R70_RUNTIME_CANDIDATE_DIGEST", "b" * 64)

    original_start = host.start

    def start_and_complete_maintenance() -> None:
        original_start()
        maintenance_root = profile / "data/topic59-maintenance-r71b"
        (profile / "cron/output" / transaction.KNOWN_R70_CRON_JOB_ID).mkdir(
            parents=True,
            exist_ok=True,
        )
        output = profile / "cron/output" / transaction.KNOWN_R70_CRON_JOB_ID / "r71b.md"
        _ = output.write_text("Status: maintenance no-op completed\n", encoding="utf-8")
        (maintenance_root / "hold.json").unlink()
        _write_audit(maintenance_root / "skip-audit.json", artifacts)
        replacement = cron.with_suffix(".next")
        _ = replacement.write_text(
            json.dumps({"jobs": [_r70_job(status="ok", error=None)]}),
            encoding="utf-8",
        )
        os.replace(replacement, cron)

    monkeypatch.setattr(host, "start", start_and_complete_maintenance)
    approval = "AUTHORIZE NUTRICOACH V1.5 R71B MAINTENANCE UPGRADE " + artifacts.final_package_digest

    # When
    _ = execute_disposable(
        approval,
        tmp_path,
        host,
        expected_approval=approval,
        maintenance_binding=binding,
    )

    # Then
    assert PhaseJournal(host.paths.execution_root / "phase.json").phase() == "COMMITTED"
    assert host.ledger_consumed()
    assert service.running
    credential = host.paths.successor_runtime.parent / "runtime-authority/nutricoach-topic59-maintenance-r71b.json"
    assert credential.read_bytes() == artifacts.authority_bytes
    assert host.verify_exact_postimages()
    assert not host.provider_events
    assert not host.network_events


def test_fresh_confirmed_recovery_loads_durable_successor_postimages(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    # Given: an actual disposable successor reaches durable confirmation before
    # process death, so the recovery host has no in-memory postimage map.
    _ = target_fixture(tmp_path)
    service = DisposableService()
    host = ConcreteLiveHost.disposable(tmp_path, service)
    artifacts = _artifacts(host.candidate_digest)
    profile = host.paths.profile
    cron = profile / "cron/jobs.json"
    cron.parent.mkdir()
    _ = cron.write_text(
        json.dumps({"jobs": [_r70_job(
            status="error", error=transaction.KNOWN_R70_CRON_ERROR,
        )]}),
        encoding="utf-8",
    )
    ledger_path = profile / "data/weekly-operations-topic59.jsonl"
    _ = ledger_path.write_bytes(b"historical sending rows\n")
    evidence = tmp_path / "task-10-cron-no-send-recovery-analysis.json"
    _ = evidence.write_bytes(b"accepted r70 exception\n")
    r70_candidate = tmp_path / "r70-candidate"
    _ = r70_candidate.write_text("b" * 64 + "\n", encoding="utf-8")
    authority = tmp_path / "maintenance-authority.json"
    _ = authority.write_bytes(artifacts.authority_bytes)
    binding: dict[str, JsonValue] = {
        "maintenance_authority_path": str(authority),
        "maintenance_authority_sha256": artifacts.authority_sha256,
        "package_binding_digest": artifacts.package_binding_digest,
        "candidate_digest": host.candidate_digest,
        "maintenance_ledger_path": str(ledger_path),
        "maintenance_ledger_sha256": hashlib.sha256(ledger_path.read_bytes()).hexdigest(),
        "maintenance_output_directory": str(
            profile / "cron/output" / transaction.KNOWN_R70_CRON_JOB_ID
        ),
        "r70_error_evidence_path": str(evidence),
        "r70_error_evidence_sha256": hashlib.sha256(evidence.read_bytes()).hexdigest(),
        "authority_baseline": {"candidate_path": str(r70_candidate)},
        "package_digest": artifacts.final_package_digest,
    }
    evidence_sha = hashlib.sha256(evidence.read_bytes()).hexdigest()
    for module in (transaction, sealed_host):
        monkeypatch.setattr(module, "KNOWN_R70_CRON_ERROR_EVIDENCE_SHA256", evidence_sha)
        monkeypatch.setattr(module, "KNOWN_R70_RUNTIME_CANDIDATE_DIGEST", "b" * 64)
    authorization = GlobalLedger(host.paths.ledger_root, host.candidate_digest)
    authorization.reserve()
    journal = PhaseJournal(host.paths.execution_root / "phase.json")
    host.capture_preflight()
    host.verify_preflight()
    host.write_recovery_manifest()
    host.stop()
    host.record_snapshot()
    host.stopped_probe()
    host.install()
    host.off_smoke()
    host.migration_dry_run()
    host.migration_apply()
    host.weekly_startup_smoke()
    host.promote_runtime_authority()
    maintenance = sealed_host.build_r71b_maintenance_controller(host, binding, journal)
    approval = (
        "AUTHORIZE NUTRICOACH V1.5 R71B MAINTENANCE UPGRADE "
        + artifacts.final_package_digest
    )
    maintenance.arm(approval)
    host.switch_systemd()
    host.reload()
    original_start = host.start

    def start_and_confirm() -> None:
        original_start()
        root = profile / "data/topic59-maintenance-r71b"
        output = profile / "cron/output" / transaction.KNOWN_R70_CRON_JOB_ID / "r71b.md"
        output.parent.mkdir(parents=True, exist_ok=True)
        _ = output.write_text("Status: maintenance no-op completed\n", encoding="utf-8")
        (root / "hold.json").unlink()
        _write_audit(root / "skip-audit.json", artifacts)
        replacement = cron.with_suffix(".next")
        _ = replacement.write_text(
            json.dumps({"jobs": [_r70_job(status="ok", error=None)]}),
            encoding="utf-8",
        )
        os.replace(replacement, cron)

    monkeypatch.setattr(host, "start", start_and_confirm)
    maintenance.complete_start()
    assert journal.phase() == "MAINTENANCE_CONFIRMED"
    assert host.verify_exact_postimages()

    # When: a fresh host reconstructs the postimage map from disk and performs
    # the real post-fence rather than a mocked substitute.
    recovered_service = DisposableService()
    recovered = ConcreteLiveHost.disposable(tmp_path, recovered_service)
    recover_pending(recovered, binding)

    # Then
    assert journal.phase() == "COMMITTED"
    assert authorization.outcome() == "SUCCEEDED"
    assert recovered_service.running
    assert recovered.verify_exact_postimages()
    assert not recovered.provider_events
    assert not recovered.network_events


def test_preconfirmation_start_fault_restores_and_pauses_the_predecessor(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    # Given
    paths = target_fixture(tmp_path)
    service = DisposableService()
    host = ConcreteLiveHost.disposable(
        tmp_path,
        service,
        fault=RuntimeError("start"),
        fault_stage="start",
    )
    artifacts = _artifacts(host.candidate_digest)
    profile = host.paths.profile
    cron = profile / "cron/jobs.json"
    cron.parent.mkdir()
    _ = cron.write_text(
        json.dumps({"jobs": [_r70_job(
            status="error", error=transaction.KNOWN_R70_CRON_ERROR,
        )]}),
        encoding="utf-8",
    )
    before = tuple(path.read_bytes() for path in paths)
    ledger_path = profile / "data/weekly-operations-topic59.jsonl"
    _ = ledger_path.write_bytes(b"historical sending rows\n")
    evidence = tmp_path / "task-10-cron-no-send-recovery-analysis.json"
    _ = evidence.write_bytes(b"accepted r70 exception\n")
    r70_candidate = tmp_path / "r70-candidate"
    _ = r70_candidate.write_text("b" * 64 + "\n", encoding="utf-8")
    authority = tmp_path / "maintenance-authority.json"
    _ = authority.write_bytes(artifacts.authority_bytes)
    binding: dict[str, JsonValue] = {
        "maintenance_authority_path": str(authority),
        "maintenance_authority_sha256": artifacts.authority_sha256,
        "package_binding_digest": artifacts.package_binding_digest,
        "candidate_digest": host.candidate_digest,
        "maintenance_ledger_path": str(ledger_path),
        "maintenance_ledger_sha256": hashlib.sha256(ledger_path.read_bytes()).hexdigest(),
        "maintenance_output_directory": str(
            profile / "cron/output" / transaction.KNOWN_R70_CRON_JOB_ID
        ),
        "r70_error_evidence_path": str(evidence),
        "r70_error_evidence_sha256": hashlib.sha256(evidence.read_bytes()).hexdigest(),
        "authority_baseline": {"candidate_path": str(r70_candidate)},
        "package_digest": artifacts.final_package_digest,
    }
    evidence_sha = hashlib.sha256(evidence.read_bytes()).hexdigest()
    for module in (transaction, sealed_host):
        monkeypatch.setattr(module, "KNOWN_R70_CRON_ERROR_EVIDENCE_SHA256", evidence_sha)
        monkeypatch.setattr(module, "KNOWN_R70_RUNTIME_CANDIDATE_DIGEST", "b" * 64)
    approval = (
        "AUTHORIZE NUTRICOACH V1.5 R71B MAINTENANCE UPGRADE "
        + artifacts.final_package_digest
    )

    # When / Then
    with pytest.raises(RuntimeError, match="start") as failure:
        _ = execute_disposable(
            approval,
            tmp_path,
            host,
            expected_approval=approval,
            maintenance_binding=binding,
        )
    assert not getattr(failure.value, "__notes__", ()), failure.value.__notes__

    cron_document = _OBJECT.validate_json(cron.read_bytes())
    jobs = cron_document.get("jobs")
    assert isinstance(jobs, list) and len(jobs) == 1 and isinstance(jobs[0], dict)
    paused = jobs[0]
    assert paused["enabled"] is False
    assert paused["state"] == "paused"
    assert paused["last_error"] == transaction.KNOWN_R70_CRON_ERROR
    assert PhaseJournal(host.paths.execution_root / "phase.json").phase() == (
        "ROLLED_BACK_SAFE_CRON_PAUSED"
    )
    assert service.running
    assert service.observe()["NRestarts"] == "0"
    assert tuple(path.read_bytes() for path in paths) == before
    assert not host.provider_events
    assert not host.network_events