from __future__ import annotations

import importlib.util
import json
import stat
import subprocess
import sys
from types import SimpleNamespace
from pathlib import Path
from typing import Any

import pytest

ROOT = Path(__file__).resolve().parents[3]
CONTROLLER = ROOT / "scripts" / "backup" / "signoz-metadata-isolated-restore.py"
POLICY = ROOT / "config" / "signoz" / "metadata-export-policy.json"
TRACE_ID = "trace-P0-OBS-002-isolated"
RUN_ID = "run-P0-OBS-002-isolated"
WORK_ITEM_ID = "P0-OBS-002"


def load_controller() -> Any:
    backup_dir = str(ROOT / "scripts" / "backup")
    if backup_dir not in sys.path:
        sys.path.insert(0, backup_dir)
    spec = importlib.util.spec_from_file_location("signoz_isolated_restore", CONTROLLER)
    assert spec is not None and spec.loader is not None
    module = importlib.util.module_from_spec(spec)
    sys.modules[spec.name] = module
    spec.loader.exec_module(module)
    return module


@pytest.fixture()
def controller() -> Any:
    return load_controller()


class RecordingRunner:
    def __init__(self, controller: Any) -> None:
        self.controller = controller
        self.commands: list[list[str]] = []
        self.resources: dict[tuple[str, str], dict[str, Any]] = {}
        self.production = {
            name: {
                "Id": (name.encode().hex() + "0" * 64)[:64],
                "Image": image_id,
                "Config": {"Image": ref},
                "State": {
                    "Running": True,
                    "StartedAt": "2026-07-22T00:00:00Z",
                },
                "RestartCount": 0,
            }
            for name, (ref, image_id) in controller.EXPECTED_PRODUCTION.items()
        }
        self.image_rows = ["sha256:test|test:fixture"]

    def __call__(self, argv: list[str], _timeout: int) -> Any:
        self.commands.append(argv[:])
        c = self.controller
        if argv[1:3] == ["image", "inspect"]:
            ref = argv[-1]
            for expected_ref, image_id in c.EXPECTED_IMAGES.values():
                if ref == expected_ref:
                    return c.CommandResult(0, f"{image_id}\n")
            return c.CommandResult(1, "")
        if argv[1:3] == ["image", "ls"]:
            return c.CommandResult(0, "\n".join(self.image_rows) + "\n")
        if argv[1:2] == ["info"]:
            return c.CommandResult(0, "28.0.0\n")
        if len(argv) >= 4 and argv[2] == "inspect":
            kind, name = argv[1], argv[3]
            if kind == "container" and name in self.production:
                return c.CommandResult(0, json.dumps([self.production[name]]))
            value = self.resources.get((kind, name))
            if value is None:
                return c.CommandResult(1, "")
            return c.CommandResult(0, json.dumps([value]))
        if argv[1:3] == ["network", "create"]:
            name = argv[-1]
            labels = labels_from_command(argv)
            self.resources[("network", name)] = {"Name": name, "Labels": labels}
            return c.CommandResult(0, f"{name}\n")
        if argv[1:3] == ["volume", "create"]:
            name = argv[-1]
            labels = labels_from_command(argv)
            self.resources[("volume", name)] = {"Name": name, "Labels": labels}
            return c.CommandResult(0, f"{name}\n")
        if argv[1:2] == ["run"]:
            name = argv[argv.index("--name") + 1]
            ref = argv[-1]
            role = next(
                candidate
                for candidate, (expected_ref, _image_id) in c.EXPECTED_IMAGES.items()
                if expected_ref == ref
            )
            limits = c.RESOURCE_LIMITS[role]
            expected_id = next(
                image_id
                for expected_ref, image_id in c.EXPECTED_IMAGES.values()
                if expected_ref == ref
            )
            self.resources[("container", name)] = {
                "Id": (name.encode().hex() + "0" * 64)[:64],
                "Image": expected_id,
                "Config": {"Image": ref, "Labels": labels_from_command(argv)},
                "HostConfig": {
                    "RestartPolicy": {"Name": "no"},
                    "NanoCpus": limits["nano_cpus"],
                    "Memory": limits["memory_bytes"],
                    "MemorySwap": limits["memory_swap_bytes"],
                    "PidsLimit": limits["pids_limit"],
                    "LogConfig": {
                        "Type": c.RESOURCE_LOG_CONFIG["type"],
                        "Config": {
                            "max-file": c.RESOURCE_LOG_CONFIG["max_file"],
                            "max-size": c.RESOURCE_LOG_CONFIG["max_size"],
                        },
                    },
                },
                "State": {"Running": True},
                "NetworkSettings": {
                    "Ports": (
                        {"8080/tcp": [{"HostIp": "127.0.0.1", "HostPort": "49173"}]}
                        if name.endswith("-server")
                        else {}
                    )
                },
                "RestartCount": 0,
            }
            return c.CommandResult(0, f"{name}\n")
        if argv[1:2] == ["port"]:
            return c.CommandResult(0, "127.0.0.1:49173\n")
        if argv[1:2] == ["exec"] and argv[-2:] == ["--query", "SELECT 1"]:
            return c.CommandResult(0, "1\n")
        if argv[1:3] == ["rm", "--force"]:
            self.resources.pop(("container", argv[-1]), None)
            return c.CommandResult(0, "")
        if argv[1:3] == ["volume", "rm"]:
            self.resources.pop(("volume", argv[-1]), None)
            return c.CommandResult(0, "")
        if argv[1:3] == ["network", "rm"]:
            self.resources.pop(("network", argv[-1]), None)
            return c.CommandResult(0, "")
        if argv[1:2] == ["ps"] or (
            len(argv) >= 3 and argv[1] in {"volume", "network"} and argv[2] == "ls"
        ):
            label = argv[-1].removeprefix("label=")
            key, expected = label.split("=", 1)
            matches = []
            for (kind, name), value in self.resources.items():
                if argv[1] == "ps" and kind != "container":
                    continue
                if argv[1] in {"volume", "network"} and kind != argv[1]:
                    continue
                labels = (
                    value.get("Config", {}).get("Labels", {})
                    if kind == "container"
                    else value.get("Labels", {})
                )
                if labels.get(key) == expected:
                    matches.append(name)
            return c.CommandResult(0, "\n".join(matches))
        return c.CommandResult(1, "")


def labels_from_command(argv: list[str]) -> dict[str, str]:
    labels: dict[str, str] = {}
    for index, value in enumerate(argv):
        if value == "--label":
            key, label_value = argv[index + 1].split("=", 1)
            labels[key] = label_value
    return labels


def test_identity_and_exact_resource_names(controller: Any) -> None:
    plan = controller.build_resource_plan(TRACE_ID, RUN_ID, WORK_ITEM_ID)

    assert plan.identity == controller.derive_identity(TRACE_ID, RUN_ID, WORK_ITEM_ID)
    assert len(plan.identity) == 64
    assert plan.prefix == f"awoooi-signoz-md-restore-{plan.identity[:16]}"
    assert plan.canonical_id == (
        f"ephemeral-signoz-metadata-restore-{plan.identity[:16]}"
    )
    assert plan.containers == (
        f"{plan.prefix}-server",
        f"{plan.prefix}-clickhouse",
        f"{plan.prefix}-zookeeper",
    )
    assert len(plan.volumes) == 4
    assert plan.workspace == Path(f"/tmp/{plan.prefix}-private")


def test_policy_requires_reviewed_exact_startup_and_images(controller: Any) -> None:
    policy = json.loads(POLICY.read_text(encoding="utf-8"))
    isolation = controller.validate_restore_policy(policy)
    assert isolation["startup_contract"]["reviewed"] is True
    assert isolation["resource_limits"] == controller.RESOURCE_LIMITS
    assert isolation["resource_log_config"] == controller.RESOURCE_LOG_CONFIG

    policy["restore_isolation"]["startup_contract"]["reviewed"] = False
    with pytest.raises(
        controller.ContractError, match="restore_startup_contract_not_reviewed"
    ):
        controller.validate_restore_policy(policy)

    policy = json.loads(POLICY.read_text(encoding="utf-8"))
    policy["restore_isolation"]["images"]["clickhouse"]["id"] = f"sha256:{'a' * 64}"
    with pytest.raises(
        controller.ContractError, match="restore_image_policy_not_exact"
    ):
        controller.validate_restore_policy(policy)

    policy = json.loads(POLICY.read_text(encoding="utf-8"))
    policy["assets"][0]["classification"] = "export_only"
    with pytest.raises(
        controller.ContractError, match="restore_portable_asset_ids_not_exact"
    ):
        controller.validate_restore_policy(policy)


def test_resource_creation_is_internal_no_pull_no_restart_and_loopback(
    controller: Any,
) -> None:
    runner = RecordingRunner(controller)
    plan = controller.build_resource_plan(TRACE_ID, RUN_ID, WORK_ITEM_ID)
    cluster = ROOT / "ops/signoz/clickhouse/restore-drill/config.d/isolated-cluster.xml"

    controller.create_resources(runner, plan, RUN_ID, cluster)

    commands = runner.commands
    network = next(
        command for command in commands if command[1:3] == ["network", "create"]
    )
    assert "--internal" in network
    run_commands = [command for command in commands if command[1:2] == ["run"]]
    assert len(run_commands) == 3
    for command in run_commands:
        assert "--pull=never" in command
        assert "--restart=no" in command
        assert f"{controller.PRIMARY_LABEL}={RUN_ID}" in command
        assert f"{controller.IDENTITY_LABEL}={plan.identity}" in command
        assert not any("signoz-clickhouse:" in value for value in command)
        ref = command[-1]
        role = next(
            candidate
            for candidate, (
                expected_ref,
                _image_id,
            ) in controller.EXPECTED_IMAGES.items()
            if expected_ref == ref
        )
        limits = controller.RESOURCE_LIMITS[role]
        assert command[command.index("--cpus") + 1] == limits["cpus"]
        assert command[command.index("--memory") + 1] == limits["memory"]
        assert command[command.index("--memory-swap") + 1] == limits["memory_swap"]
        assert command[command.index("--pids-limit") + 1] == str(limits["pids_limit"])
        assert command[command.index("--log-driver") + 1] == "local"
        assert "max-size=10m" in command
        assert "max-file=2" in command
    server = next(
        command for command in run_commands if plan.server_container in command
    )
    assert "127.0.0.1::8080" in server
    assert f"{plan.server_data_volume}:/var/lib/signoz" in server
    assert "SIGNOZ_TELEMETRYSTORE_CLICKHOUSE_DSN=tcp://clickhouse:9000" in server
    clickhouse_run_index = commands.index(
        next(
            command for command in run_commands if plan.clickhouse_container in command
        )
    )
    readiness_index = next(
        index
        for index, command in enumerate(commands)
        if command[1:2] == ["exec"] and command[-2:] == ["--query", "SELECT 1"]
    )
    server_run_index = commands.index(server)
    assert clickhouse_run_index < readiness_index < server_run_index
    controller.verify_runtime_container_images(runner, plan)

    server_runtime = runner.resources[("container", plan.server_container)]
    server_runtime["HostConfig"]["Memory"] += 1
    with pytest.raises(
        controller.ContractError,
        match="restore_runtime_container_image_restart_or_resource_limit_drift",
    ):
        controller.verify_runtime_container_images(runner, plan)


def test_exact_image_drift_fails_before_resource_creation(controller: Any) -> None:
    runner = RecordingRunner(controller)

    def drift(argv: list[str], timeout: int) -> Any:
        result = runner(argv, timeout)
        if argv[-1] == controller.EXPECTED_IMAGES["clickhouse"][0]:
            return controller.CommandResult(0, f"sha256:{'b' * 64}\n")
        return result

    with pytest.raises(
        controller.ContractError, match="restore_image_id_mismatch_clickhouse"
    ):
        controller.verify_exact_images(drift)
    assert not runner.resources


def test_cleanup_refuses_same_name_with_wrong_ownership(controller: Any) -> None:
    runner = RecordingRunner(controller)
    plan = controller.build_resource_plan(TRACE_ID, RUN_ID, WORK_ITEM_ID)
    runner.resources[("container", plan.server_container)] = {
        "Config": {
            "Labels": {
                controller.PRIMARY_LABEL: "some-other-run",
                controller.IDENTITY_LABEL: plan.identity,
            }
        }
    }

    with pytest.raises(
        controller.ContractError, match="cleanup_container_ownership_mismatch"
    ):
        controller.remove_owned_resource(
            runner,
            plan,
            RUN_ID,
            "container",
            plan.server_container,
        )
    assert ("container", plan.server_container) in runner.resources
    assert not any(command[1:3] == ["rm", "--force"] for command in runner.commands)


def test_cleanup_removes_exact_resources_and_proves_continuity(
    controller: Any,
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    runner = RecordingRunner(controller)
    plan = controller.build_resource_plan(
        TRACE_ID, RUN_ID, WORK_ITEM_ID, workspace_parent=tmp_path
    )
    cluster = ROOT / "ops/signoz/clickhouse/restore-drill/config.d/isolated-cluster.xml"
    controller.create_workspace(plan)
    controller.create_resources(runner, plan, RUN_ID, cluster)
    state = {
        "production_before": controller.production_snapshot(runner),
        "image_inventory_before": controller.image_inventory(runner),
        "listener": {"host": "127.0.0.1", "port": 49173},
    }
    monkeypatch.setattr(controller, "tcp_connectable", lambda _host, _port: False)
    monkeypatch.setattr(controller.time, "sleep", lambda _seconds: None)

    errors = controller.cleanup_and_verify(runner, plan, RUN_ID, state)

    assert errors == []
    assert not runner.resources
    assert not plan.workspace.exists()


@pytest.mark.parametrize("server_created", [False, True])
def test_cleanup_reconciles_without_listener_state_after_interruption(
    controller: Any,
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
    server_created: bool,
) -> None:
    runner = RecordingRunner(controller)
    plan = controller.build_resource_plan(
        TRACE_ID, RUN_ID, WORK_ITEM_ID, workspace_parent=tmp_path
    )
    controller.create_workspace(plan)
    if server_created:
        cluster = (
            ROOT / "ops/signoz/clickhouse/restore-drill/config.d/isolated-cluster.xml"
        )
        controller.create_resources(runner, plan, RUN_ID, cluster)
    state = {
        "production_before": controller.production_snapshot(runner),
        "image_inventory_before": controller.image_inventory(runner),
        "listener": None,
    }
    monkeypatch.setattr(controller, "tcp_connectable", lambda _host, _port: False)
    monkeypatch.setattr(controller.time, "sleep", lambda _seconds: None)

    errors = controller.cleanup_and_verify(runner, plan, RUN_ID, state)

    assert errors == []
    assert not runner.resources
    assert not plan.workspace.exists()


def test_cleanup_already_clean_without_state_uses_fresh_continuity(
    controller: Any,
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    runner = RecordingRunner(controller)
    plan = controller.build_resource_plan(
        TRACE_ID, RUN_ID, WORK_ITEM_ID, workspace_parent=tmp_path
    )
    monkeypatch.setattr(controller, "tcp_connectable", lambda _host, _port: False)
    monkeypatch.setattr(controller.time, "sleep", lambda _seconds: None)

    errors = controller.cleanup_and_verify(runner, plan, RUN_ID, None)

    assert errors == []
    assert not plan.workspace.exists()


def test_cleanup_reconciles_resource_materialized_after_first_inventory(
    controller: Any,
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    runner = RecordingRunner(controller)
    plan = controller.build_resource_plan(
        TRACE_ID, RUN_ID, WORK_ITEM_ID, workspace_parent=tmp_path
    )
    sleeps = 0

    def materialize_after_first_pass(_seconds: float) -> None:
        nonlocal sleeps
        sleeps += 1
        if sleeps == 1:
            runner.resources[("volume", plan.server_data_volume)] = {
                "Name": plan.server_data_volume,
                "Labels": {
                    controller.PRIMARY_LABEL: RUN_ID,
                    controller.IDENTITY_LABEL: plan.identity,
                },
            }

    monkeypatch.setattr(controller, "tcp_connectable", lambda _host, _port: False)
    monkeypatch.setattr(controller.time, "sleep", materialize_after_first_pass)

    errors = controller.cleanup_and_verify(runner, plan, RUN_ID, None)

    assert errors == []
    assert not runner.resources
    removals = [
        command
        for command in runner.commands
        if command[1:3] == ["volume", "rm"] and command[-1] == plan.server_data_volume
    ]
    assert len(removals) == 1


def test_apply_cli_requires_durable_receipt_before_runtime_dispatch() -> None:
    result = subprocess.run(
        [
            sys.executable,
            str(CONTROLLER),
            "--apply",
            "--trace-id",
            TRACE_ID,
            "--run-id",
            RUN_ID,
            "--work-item-id",
            WORK_ITEM_ID,
        ],
        text=True,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        timeout=10,
        check=False,
    )

    assert result.returncode == 2
    assert "receipt_file_required_for_restore_apply" in result.stderr
    assert result.stdout == ""


def test_distinct_verifier_requires_immutable_apply_receipt(
    controller: Any,
    tmp_path: Path,
) -> None:
    args = SimpleNamespace(
        trace_id=TRACE_ID,
        run_id=RUN_ID,
        work_item_id=WORK_ITEM_ID,
    )
    plan = controller.build_resource_plan(
        TRACE_ID, RUN_ID, WORK_ITEM_ID, workspace_parent=tmp_path
    )
    executor_receipt = controller.executor_success(
        args,
        plan,
        {
            "restoreSemanticDigest": "a" * 64,
            "restoredAssetCounts": {
                "dashboards": 1,
                "alert_rules": 2,
                "explorer_views_logs": 3,
                "explorer_views_meter": 0,
                "explorer_views_metrics": 0,
                "explorer_views_traces": 0,
            },
            "listenerBinding": {"host": "127.0.0.1", "port": 49173},
            "productionContinuitySha256": "b" * 64,
            "imageInventorySha256": "c" * 64,
            "exportBundleManifestSha256": "d" * 64,
        },
    )
    path = tmp_path / "apply.json"
    controller.write_terminal(path, executor_receipt)

    validated = controller.validate_apply_receipt(path, args, plan)
    verified = controller.independent_success(args, plan, validated)

    assert executor_receipt["completionClaim"] is False
    assert executor_receipt["executorZeroResidueVerified"] is True
    assert executor_receipt["portableAssetIds"] == [
        "dashboards",
        "alert_rules",
        "explorer_views_logs",
        "explorer_views_meter",
        "explorer_views_metrics",
        "explorer_views_traces",
    ]
    assert verified["phase"] == "independent_restore_verifier"
    assert verified["completionClaim"] is True
    assert verified["independentZeroResidueVerified"] is True
    assert verified["restoreSemanticDigest"] == "a" * 64
    assert verified["applyReceiptValidated"] is True
    assert len(verified["applyReceiptSha256"]) == 64
    assert verified["listenerBinding"] == {
        "host": "127.0.0.1",
        "port": 49173,
    }
    assert verified["productionContinuityVerified"] is True
    assert verified["imageInventoryContinuityVerified"] is True
    assert verified["listenerBindingUnreachableVerified"] is True

    stale_receipt = json.loads(json.dumps(executor_receipt))
    del stale_receipt["restoredAssetCounts"]["explorer_views_meter"]
    stale_path = tmp_path / "stale-apply.json"
    controller.write_terminal(stale_path, stale_receipt)
    with pytest.raises(
        controller.ContractError,
        match="apply_receipt_semantic_evidence_invalid",
    ):
        controller.validate_apply_receipt(stale_path, args, plan)


def test_independent_cleanup_uses_apply_listener_and_continuity_after_workspace_gone(
    controller: Any,
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    runner = RecordingRunner(controller)
    plan = controller.build_resource_plan(
        TRACE_ID, RUN_ID, WORK_ITEM_ID, workspace_parent=tmp_path
    )
    production_digest = controller.sha256_bytes(
        controller.canonical_json_bytes(controller.production_snapshot(runner))
    )
    inventory_digest = controller.sha256_bytes(
        controller.canonical_json_bytes(controller.image_inventory(runner))
    )
    continuity = {
        "listenerBinding": {"host": "127.0.0.1", "port": 49173},
        "productionContinuitySha256": production_digest,
        "imageInventorySha256": inventory_digest,
    }
    probed: list[tuple[str, int]] = []

    def probe(host: str, port: int) -> bool:
        probed.append((host, port))
        return False

    monkeypatch.setattr(controller, "tcp_connectable", probe)
    monkeypatch.setattr(controller.time, "sleep", lambda _seconds: None)

    errors = controller.cleanup_and_verify(
        runner,
        plan,
        RUN_ID,
        None,
        continuity,
    )

    assert errors == []
    assert probed == [("127.0.0.1", 49173)] * controller.CLEANUP_STABLE_PASSES
    assert not plan.workspace.exists()


def test_restore_driver_apply_uses_and_validates_private_child_receipt(
    controller: Any,
    tmp_path: Path,
) -> None:
    args = SimpleNamespace(
        bundle_dir=tmp_path / "bundle",
        policy=POLICY,
        trace_id=TRACE_ID,
        run_id=RUN_ID,
        work_item_id=WORK_ITEM_ID,
    )
    isolation_path = tmp_path / "isolation-receipt.json"
    isolation_path.write_text("{}", encoding="utf-8")
    token_path = tmp_path / "session-access-token"
    token_path.write_text("test-only-token\n", encoding="ascii")
    commands: list[list[str]] = []

    def child_runner(argv: list[str], _timeout: int) -> Any:
        commands.append(argv[:])
        receipt_path = Path(argv[argv.index("--receipt-file") + 1])
        value = controller.receipt(
            trace_id=TRACE_ID,
            run_id=RUN_ID,
            work_item_id=WORK_ITEM_ID,
            phase="terminal",
            terminal="partial_degraded_cleanup_pending",
            detail="isolated_restore_actions_3_semantic_readback_pass_cleanup_pending",
        )
        value.update(
            {
                "restored_asset_counts": {
                    "dashboards": 1,
                    "alert_rules": 1,
                    "explorer_views_logs": 1,
                    "explorer_views_meter": 0,
                    "explorer_views_metrics": 0,
                    "explorer_views_traces": 0,
                },
                "restored_asset_semantic_sha256": {
                    asset_id: "a" * 64 for asset_id in controller.PORTABLE_ASSET_IDS
                },
                "portable_semantic_sha256": "b" * 64,
                "completion_scope": "portable_assets_only",
                "completion_claim": False,
                "full_sqlite_completion_claim": False,
                "roles_preferences_global_config_restore_claim": False,
                "secret_bearing_assets_restore_claim": False,
            }
        )
        receipt_path.write_bytes(controller.canonical_json_bytes(value))
        receipt_path.chmod(0o600)
        return controller.CommandResult(
            0,
            json.dumps(value, sort_keys=True, separators=(",", ":")),
        )

    result = controller.invoke_restore_driver(
        args,
        child_runner,
        "http://127.0.0.1:49173",
        isolation_path,
        token_path,
        "--apply",
    )

    assert result["portable_semantic_sha256"] == "b" * 64
    assert len(commands) == 1
    assert "--receipt-file" in commands[0]
    child_path = Path(commands[0][commands[0].index("--receipt-file") + 1])
    assert child_path == tmp_path / "restore-driver-apply-receipt.json"
    assert stat.S_IMODE(child_path.stat().st_mode) == 0o600


def test_interrupted_verify_cleanup_reconciles_again_before_zero_residue_receipt(
    controller: Any,
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    plan = controller.build_resource_plan(
        TRACE_ID, RUN_ID, WORK_ITEM_ID, workspace_parent=tmp_path
    )
    calls = 0
    emitted: list[dict[str, Any]] = []

    def interrupt_once(
        _runner: Any,
        _plan: Any,
        _run_id: str,
        _state: dict[str, Any] | None,
        _continuity: dict[str, Any] | None = None,
    ) -> list[str]:
        nonlocal calls
        calls += 1
        if calls == 1:
            raise controller.SignalAbort("signal_15_cleanup_required")
        return []

    monkeypatch.setattr(
        sys,
        "argv",
        [
            str(CONTROLLER),
            "--verify-cleanup",
            "--trace-id",
            TRACE_ID,
            "--run-id",
            RUN_ID,
            "--work-item-id",
            WORK_ITEM_ID,
        ],
    )
    monkeypatch.setattr(controller, "build_resource_plan", lambda *_args: plan)
    monkeypatch.setattr(controller, "load_policy", lambda _path: {})
    monkeypatch.setattr(controller, "load_state_if_safe", lambda _plan: None)
    monkeypatch.setattr(controller, "cleanup_and_verify", interrupt_once)
    monkeypatch.setattr(controller, "emit", emitted.append)

    exit_code = controller.main()

    assert exit_code == 1
    assert calls == 2
    assert emitted[-1]["terminal"] == "failed_zero_residue_verified_no_completion"
    assert emitted[-1]["zeroResidueVerified"] is True
    assert emitted[-1]["cleanupErrorCodes"] == []
    assert "signal_15_cleanup_required" in emitted[-1]["detail"]


def test_signal_at_cleanup_transition_still_reconciles_and_writes_terminal(
    controller: Any,
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    plan = controller.build_resource_plan(
        TRACE_ID, RUN_ID, WORK_ITEM_ID, workspace_parent=tmp_path
    )
    receipt_path = tmp_path / "transition-terminal.json"
    args = SimpleNamespace(
        check=False,
        apply=True,
        verify_cleanup=False,
        bundle_dir=str(tmp_path / "bundle"),
        policy=str(POLICY),
        cluster_config=str(
            ROOT / "ops/signoz/clickhouse/restore-drill/config.d/isolated-cluster.xml"
        ),
        trace_id=TRACE_ID,
        run_id=RUN_ID,
        work_item_id=WORK_ITEM_ID,
        receipt_file=str(receipt_path),
        apply_receipt_file=None,
    )
    state = {
        "production_before": {},
        "image_inventory_before": ["sha256:test|test:fixture"],
        "listener": {"host": "127.0.0.1", "port": 49173},
    }
    cleanup_calls = 0
    installed: dict[int, Any] = {}
    transition_signal_sent = False

    def fake_getsignal(signum: int) -> Any:
        return installed.get(signum, "original-handler")

    def fake_signal(signum: int, handler: Any) -> Any:
        nonlocal transition_signal_sent
        previous = installed.get(signum, "original-handler")
        installed[signum] = handler
        if (
            handler is controller.signal.SIG_IGN
            and not transition_signal_sent
            and callable(previous)
        ):
            transition_signal_sent = True
            previous(controller.signal.SIGTERM, None)
        return previous

    def cleanup_once(
        _runner: Any,
        _plan: Any,
        _run_id: str,
        _state: dict[str, Any] | None,
        _continuity: dict[str, Any] | None = None,
    ) -> list[str]:
        nonlocal cleanup_calls
        cleanup_calls += 1
        return []

    monkeypatch.setattr(controller, "parse_args", lambda: args)
    monkeypatch.setattr(controller, "build_resource_plan", lambda *_args: plan)
    monkeypatch.setattr(controller, "load_policy", lambda _path: {})
    monkeypatch.setattr(
        controller,
        "run_apply",
        lambda *_args: {
            "restoreSemanticDigest": "a" * 64,
            "restoredAssetCounts": {},
        },
    )
    monkeypatch.setattr(controller, "load_state_if_safe", lambda _plan: state)
    monkeypatch.setattr(controller, "cleanup_and_verify", cleanup_once)
    monkeypatch.setattr(controller.signal, "getsignal", fake_getsignal)
    monkeypatch.setattr(controller.signal, "signal", fake_signal)

    exit_code = controller.main()

    terminal = json.loads(receipt_path.read_text(encoding="utf-8"))
    assert exit_code == 1
    assert transition_signal_sent is True
    assert cleanup_calls == 1
    assert terminal["terminal"] == "failed_zero_residue_verified_no_completion"
    assert terminal["zeroResidueVerified"] is True
    assert terminal["cleanupErrorCodes"] == []
    assert "signal_15_cleanup_required" in terminal["detail"]


def test_bootstrap_secrets_stay_in_mode_0400_files(
    controller: Any,
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
    capsys: pytest.CaptureFixture[str],
) -> None:
    calls: list[tuple[str, dict[str, Any] | None]] = []

    def fake_request(
        url: str,
        *,
        method: str = "GET",
        payload: dict[str, Any] | None = None,
        authorization: str | None = None,
        timeout: int = 10,
    ) -> dict[str, Any]:
        assert authorization is None
        calls.append((url, payload))
        if url.endswith("/api/v1/register"):
            return {"data": {"orgId": "isolated-org"}, "status": "success"}
        return {
            "data": {
                "accessToken": "a" * 64,
                "refreshToken": "r" * 64,
            },
            "status": "success",
        }

    monkeypatch.setattr(controller, "request_json", fake_request)

    token = controller.bootstrap_isolated_session("http://127.0.0.1:49173", tmp_path)

    assert token == tmp_path / "session-access-token"
    assert stat.S_IMODE(token.stat().st_mode) == 0o400
    assert sorted(path.name for path in tmp_path.iterdir()) == ["session-access-token"]
    assert calls[0][0].endswith("/api/v1/register")
    assert calls[1][0].endswith("/api/v2/sessions/email_password")
    assert capsys.readouterr().out == ""


def test_bootstrap_accepts_live_root_response_envelope(
    controller: Any,
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    def fake_request(
        url: str,
        **_kwargs: Any,
    ) -> dict[str, Any]:
        if url.endswith("/api/v1/register"):
            return {"orgId": "isolated-org"}
        return {"accessToken": "a" * 64, "refreshToken": "r" * 64}

    monkeypatch.setattr(controller, "request_json", fake_request)

    token = controller.bootstrap_isolated_session("http://127.0.0.1:49173", tmp_path)

    assert token.read_text(encoding="ascii").strip() == "a" * 64
    assert stat.S_IMODE(token.stat().st_mode) == 0o400


def test_controller_source_has_signal_cleanup_and_no_secret_argv() -> None:
    source = CONTROLLER.read_text(encoding="utf-8")
    assert "signal.SIGHUP" in source
    assert "signal.SIGINT" in source
    assert "signal.SIGTERM" in source
    assert "finally:" in source
    assert "cleanup_and_verify(" in source
    assert '"--pull=never"' in source
    assert '"--restart=no"' in source
    assert '"--auth-mode"' in source
    assert (
        "access_token"
        not in source[
            source.index("argv = [") : source.index(
                "output = run_required", source.index("argv = [")
            )
        ]
    )
    assert (
        "password"
        not in source[
            source.index("argv = [") : source.index(
                "output = run_required", source.index("argv = [")
            )
        ]
    )
    durable_write = source.rindex("write_terminal(Path(args.receipt_file), result)")
    final_handler_restore = source.rindex("restore_signal_handlers()")
    assert durable_write < final_handler_restore
    assert "if not args.apply and not signals_held_until_terminal:" in source