feat(signoz): add isolated metadata restore controller

scripts/backup/signoz-metadata-restore-drill.py

@@ -4,21 +4,23 @@
 from __future__ import annotations
 
 import argparse
+import hashlib
 import json
 import os
+import socket
 import stat
 
 # Child commands use fixed argv without a shell.
 import subprocess  # nosec B404
 import sys
 import urllib.error
+import urllib.parse
 import urllib.request
 from pathlib import Path
 from typing import Any
 
 from signoz_metadata_contract import (
     DEFAULT_POLICY,
-    SHA256,
     ApiClient,
     ContractError,
     canonical_json_bytes,
@@ -49,6 +51,11 @@ def parse_args() -> argparse.Namespace:
     parser.add_argument("--policy", default=str(DEFAULT_POLICY))
     parser.add_argument("--target-base-url", required=True)
     parser.add_argument("--credential-file")
+    parser.add_argument(
+        "--auth-mode",
+        choices=("signoz_api_key", "isolated_session_bearer"),
+        default="signoz_api_key",
+    )
     parser.add_argument("--isolation-receipt", required=True)
     parser.add_argument("--trace-id", required=True)
     parser.add_argument("--run-id", required=True)
@@ -57,6 +64,57 @@ def parse_args() -> argparse.Namespace:
     return parser.parse_args()
 
 
+EXPECTED_IMAGES = {
+    "signoz": {
+        "ref": "signoz/signoz:v0.113.0",
+        "id": "sha256:c17991d10966aedcb0d4d83805a0baf5310f1553d90611629036fcfaee8d969b",
+    },
+    "clickhouse": {
+        "ref": "clickhouse/clickhouse-server:25.5.6",
+        "id": "sha256:8248e5926d7304400e44ecdf0c7181fbde28e6a9b1d647780eca839f34c00cc6",
+    },
+    "zookeeper": {
+        "ref": "signoz/zookeeper:3.7.1",
+        "id": "sha256:3ab0e8f032ab58f14ac1e929ac40305a4a464cf320bdfe4151d62d6922729ba0",
+    },
+}
+PRIMARY_LABEL = "awoooi.signoz.metadata.restore.run_id"
+IDENTITY_LABEL = "com.awoooi.restore.identity"
+
+
+def restore_identity(trace_id: str, run_id: str, work_item_id: str) -> str:
+    return hashlib.sha256(
+        trace_id.encode("utf-8")
+        + b"\0"
+        + run_id.encode("utf-8")
+        + b"\0"
+        + work_item_id.encode("utf-8")
+    ).hexdigest()
+
+
+def expected_resources(identity: str) -> dict[str, Any]:
+    short_id = identity[:16]
+    prefix = f"awoooi-signoz-md-restore-{short_id}"
+    return {
+        "identity": identity,
+        "prefix": prefix,
+        "canonical_id": f"ephemeral-signoz-metadata-restore-{short_id}",
+        "network": f"{prefix}-net",
+        "containers": [
+            f"{prefix}-server",
+            f"{prefix}-clickhouse",
+            f"{prefix}-zookeeper",
+        ],
+        "volumes": [
+            f"{prefix}-zk-data",
+            f"{prefix}-zk-log",
+            f"{prefix}-ch-data",
+            f"{prefix}-server-data",
+        ],
+        "workspace": f"/tmp/{prefix}-private",
+    }
+
+
 def run_independent_bundle_verifier(args: argparse.Namespace) -> None:
     command = [
         sys.executable,
@@ -122,34 +180,60 @@ def validate_isolation_receipt(
     if not isinstance(target, dict):
         raise ContractError("isolation_target_missing")
     isolation_policy = policy["restore_isolation"]
+    identity = restore_identity(args.trace_id, args.run_id, args.work_item_id)
+    resources = expected_resources(identity)
     target_url = validate_base_url(args.target_base_url, loopback_only=True)
     if target.get("base_url_sha256") != sha256_bytes(target_url.encode("utf-8")):
         raise ContractError("isolation_target_url_hash_mismatch")
-    canonical_id = target.get("canonical_id")
-    if not isinstance(canonical_id, str) or not canonical_id.startswith(
-        "ephemeral-signoz-metadata-restore-"
-    ):
+    if target.get("identity") != identity:
+        raise ContractError("isolation_target_identity_mismatch")
+    if target.get("prefix") != resources["prefix"]:
+        raise ContractError("isolation_target_prefix_mismatch")
+    if target.get("canonical_id") != resources["canonical_id"]:
         raise ContractError("isolation_target_canonical_id_invalid")
     if target.get("image_ref") != isolation_policy["required_image_ref"]:
         raise ContractError("isolation_target_image_ref_mismatch")
-    image_id = target.get("image_id")
+    if target.get("image_id") != isolation_policy["required_image_id"]:
+        raise ContractError("isolation_target_image_id_mismatch")
+    if target.get("images") != EXPECTED_IMAGES:
+        raise ContractError("isolation_target_image_set_mismatch")
+    if target.get("resources") != {
+        "network": resources["network"],
+        "containers": resources["containers"],
+        "volumes": resources["volumes"],
+        "workspace": resources["workspace"],
+    }:
+        raise ContractError("isolation_target_resource_names_mismatch")
+    listener = target.get("listener")
     if (
-        not isinstance(image_id, str)
-        or not image_id.startswith("sha256:")
-        or not SHA256.fullmatch(image_id.removeprefix("sha256:"))
+        not isinstance(listener, dict)
+        or listener.get("host") != "127.0.0.1"
+        or not isinstance(listener.get("port"), int)
+        or listener["port"] <= 0
+        or listener["port"] > 65535
     ):
-        raise ContractError("isolation_target_image_id_invalid")
+        raise ContractError("isolation_target_listener_invalid")
+    expected_url = f"http://127.0.0.1:{listener['port']}"
+    if target_url != expected_url:
+        raise ContractError("isolation_target_listener_url_mismatch")
+    if args.auth_mode != "isolated_session_bearer":
+        raise ContractError("restore_auth_mode_must_be_isolated_session_bearer")
+    if target.get("authentication_mode") != args.auth_mode:
+        raise ContractError("isolation_target_auth_mode_mismatch")
     required_values = {
         "image_present_locally": True,
         "image_pull_performed": False,
+        "image_build_performed": False,
         "production_network_attached": False,
         "production_volume_attached": False,
         "ephemeral_volumes_only": True,
         "cleanup_controller_armed": True,
         "zero_residue_verifier_armed": True,
         "network_scope": "loopback_and_internal_only",
-        "resource_label_key": isolation_policy["resource_label_key"],
+        "resource_label_key": PRIMARY_LABEL,
         "resource_label_value": args.run_id,
+        "identity_label_key": IDENTITY_LABEL,
+        "identity_label_value": identity,
     }
     for key, expected in required_values.items():
         if target.get(key) != expected:
@@ -201,9 +285,14 @@ def make_client(
         return None
     if token is None:
         raise ContractError("credential_value_unavailable")
+    if args.auth_mode == "isolated_session_bearer":
+        header_name = "Authorization"
+        token = f"Bearer {token}"
+    else:
+        header_name = policy["authentication"]["header_name"]
     return ApiClient(
         base_url=validate_base_url(args.target_base_url, loopback_only=True),
-        header_name=policy["authentication"]["header_name"],
+        header_name=header_name,
         token=token,
         timeout_seconds=policy["limits"]["request_timeout_seconds"],
         max_response_bytes=policy["limits"]["max_response_bytes"],
@@ -278,6 +367,30 @@ def run_apply(
     expected = {asset["id"]: items for asset, items in assets}
     if canonical_json_bytes(after) != canonical_json_bytes(expected):
         raise ContractError("restore_semantic_readback_mismatch")
+    restored_asset_counts: dict[str, int] = {}
+    semantic_sha256: dict[str, str] = {}
+    for asset, items in assets:
+        asset_id = asset["id"]
+        restored_items = after[asset_id]
+        expected_digest = sha256_bytes(canonical_json_bytes(items))
+        restored_digest = sha256_bytes(canonical_json_bytes(restored_items))
+        if len(restored_items) != len(items):
+            raise ContractError(f"restore_semantic_count_mismatch_{asset_id}")
+        if restored_digest != expected_digest:
+            raise ContractError(f"restore_semantic_digest_mismatch_{asset_id}")
+        restored_asset_counts[asset_id] = len(restored_items)
+        semantic_sha256[asset_id] = restored_digest
+    portable_semantic_sha256 = sha256_bytes(
+        canonical_json_bytes(
+            {
+                asset_id: {
+                    "count": restored_asset_counts[asset_id],
+                    "sha256": semantic_sha256[asset_id],
+                }
+                for asset_id in sorted(restored_asset_counts)
+            }
+        )
+    )
     terminal = receipt(
         trace_id=args.trace_id,
         run_id=args.run_id,
@@ -289,6 +402,14 @@ def run_apply(
             "cleanup_and_zero_residue_verifier_pending"
         ),
     )
+    terminal["restored_asset_counts"] = restored_asset_counts
+    terminal["restored_asset_semantic_sha256"] = semantic_sha256
+    terminal["portable_semantic_sha256"] = portable_semantic_sha256
+    terminal["completion_scope"] = "portable_assets_only"
+    terminal["completion_claim"] = False
+    terminal["full_sqlite_completion_claim"] = False
+    terminal["roles_preferences_global_config_restore_claim"] = False
+    terminal["secret_bearing_assets_restore_claim"] = False
     terminal["phase_receipts"] = [
         receipt(
             trace_id=args.trace_id,
@@ -397,41 +518,153 @@ def docker_label_count(resource: str, label: str) -> int:
     return len([line for line in result.stdout.splitlines() if line.strip()])
 
 
-def require_target_unreachable(base_url: str) -> None:
-    base_url = validate_base_url(base_url, loopback_only=True)
-    request = urllib.request.Request(base_url, method="GET")
+def docker_inspect(resource: str, name: str) -> dict[str, Any] | None:
+    command = ["docker", resource, "inspect", name]
     try:
-        # base_url is validated as loopback HTTP(S) immediately above.
-        with urllib.request.urlopen(  # nosec B310
-            request, timeout=2
-        ):
-            pass
-    except urllib.error.HTTPError as exc:
-        raise ContractError("cleanup_target_listener_still_present") from exc
-    except (urllib.error.URLError, TimeoutError, OSError):
+        result = subprocess.run(  # nosec B603
+            command,
+            text=True,
+            stdout=subprocess.PIPE,
+            stderr=subprocess.DEVNULL,
+            timeout=15,
+            check=False,
+        )
+    except (OSError, subprocess.TimeoutExpired) as exc:
+        raise ContractError(f"cleanup_{resource}_inventory_failed") from exc
+    if result.returncode != 0:
+        return None
+    try:
+        value = json.loads(result.stdout)
+    except json.JSONDecodeError as exc:
+        raise ContractError(f"cleanup_{resource}_inventory_invalid") from exc
+    if not isinstance(value, list) or len(value) != 1 or not isinstance(value[0], dict):
+        raise ContractError(f"cleanup_{resource}_inventory_invalid")
+    return value[0]
+
+
+def require_target_tcp_unreachable(base_url: str) -> None:
+    base_url = validate_base_url(base_url, loopback_only=True)
+    parsed = urllib.parse.urlsplit(base_url)
+    if parsed.hostname != "127.0.0.1" or parsed.port is None:
+        raise ContractError("cleanup_target_listener_invalid")
+    try:
+        with socket.create_connection((parsed.hostname, parsed.port), timeout=2):
+            raise ContractError("cleanup_target_tcp_listener_still_present")
+    except ContractError:
+        raise
+    except OSError:
         return
-    raise ContractError("cleanup_target_listener_still_present")
+
+
+def cleanup_production_snapshot() -> dict[str, dict[str, Any]]:
+    expected = {
+        "signoz": EXPECTED_IMAGES["signoz"],
+        "signoz-clickhouse": EXPECTED_IMAGES["clickhouse"],
+        "signoz-zookeeper-1": EXPECTED_IMAGES["zookeeper"],
+        "signoz-otel-collector": {
+            "ref": "signoz/signoz-otel-collector:v0.144.1",
+            "id": "sha256:47cbed488288df7bac6aab3665275bec6d6e24fe5c545c66914827a891e538d8",
+        },
+    }
+    snapshot: dict[str, dict[str, Any]] = {}
+    for name, image in expected.items():
+        data = docker_inspect("container", name)
+        if data is None:
+            raise ContractError("cleanup_production_container_missing")
+        config = data.get("Config")
+        state = data.get("State")
+        if not isinstance(config, dict) or not isinstance(state, dict):
+            raise ContractError("cleanup_production_container_shape_invalid")
+        row = {
+            "id": data.get("Id"),
+            "image_id": data.get("Image"),
+            "image_ref": config.get("Image"),
+            "running": state.get("Running"),
+            "started_at": state.get("StartedAt"),
+            "restart_count": data.get("RestartCount"),
+        }
+        if row["image_id"] != image["id"] or row["image_ref"] != image["ref"]:
+            raise ContractError("cleanup_production_image_identity_invalid")
+        snapshot[name] = row
+    return snapshot
+
+
+def cleanup_image_inventory() -> list[str]:
+    try:
+        result = subprocess.run(  # nosec B603
+            [
+                "docker",
+                "image",
+                "ls",
+                "--no-trunc",
+                "--format={{.ID}}|{{.Repository}}:{{.Tag}}",
+            ],
+            text=True,
+            stdout=subprocess.PIPE,
+            stderr=subprocess.DEVNULL,
+            timeout=15,
+            check=False,
+        )
+    except (OSError, subprocess.TimeoutExpired) as exc:
+        raise ContractError("cleanup_image_inventory_failed") from exc
+    if result.returncode != 0:
+        raise ContractError("cleanup_image_inventory_failed")
+    rows = sorted(line for line in result.stdout.splitlines() if line.strip())
+    if not rows:
+        raise ContractError("cleanup_image_inventory_empty")
+    return rows
 
 
 def run_cleanup_verifier(
     args: argparse.Namespace,
     policy: dict[str, Any],
+    isolation: dict[str, Any],
 ) -> dict[str, Any]:
-    label = f"{policy['restore_isolation']['resource_label_key']}={args.run_id}"
-    counts = {
-        resource: docker_label_count(resource, label)
-        for resource in ("container", "volume", "network")
-    }
-    if any(counts.values()):
-        raise ContractError("cleanup_resource_residue_present")
-    require_target_unreachable(args.target_base_url)
+    identity = restore_identity(args.trace_id, args.run_id, args.work_item_id)
+    resources = expected_resources(identity)
+    for name in resources["containers"]:
+        if docker_inspect("container", name) is not None:
+            raise ContractError("cleanup_exact_container_residue_present")
+    for name in resources["volumes"]:
+        if docker_inspect("volume", name) is not None:
+            raise ContractError("cleanup_exact_volume_residue_present")
+    if docker_inspect("network", resources["network"]) is not None:
+        raise ContractError("cleanup_exact_network_residue_present")
+    labels = (
+        f"{PRIMARY_LABEL}={args.run_id}",
+        f"{IDENTITY_LABEL}={identity}",
+    )
+    for label in labels:
+        counts = {
+            resource: docker_label_count(resource, label)
+            for resource in ("container", "volume", "network")
+        }
+        if any(counts.values()):
+            raise ContractError("cleanup_resource_label_residue_present")
+    require_target_tcp_unreachable(args.target_base_url)
+    workspace = Path(resources["workspace"])
+    if workspace.exists() or workspace.is_symlink():
+        raise ContractError("cleanup_private_workspace_residue_present")
+    production_before = isolation.get("production_before")
+    if not isinstance(production_before, dict):
+        raise ContractError("cleanup_production_before_missing")
+    if cleanup_production_snapshot() != production_before:
+        raise ContractError("cleanup_production_identity_continuity_failed")
+    image_inventory_before = isolation.get("image_inventory_before")
+    if not isinstance(image_inventory_before, list):
+        raise ContractError("cleanup_image_inventory_before_missing")
+    if cleanup_image_inventory() != image_inventory_before:
+        raise ContractError("cleanup_image_inventory_drift")
     return receipt(
         trace_id=args.trace_id,
         run_id=args.run_id,
         work_item_id=args.work_item_id,
         phase="terminal",
         terminal="pass_zero_residue_verified",
-        detail="container_0_volume_0_network_0_listener_0",
+        detail=(
+            "exact_container_0_volume_0_network_0_label_inventory_0_"
+            "tcp_listener_0_workspace_0_production_identity_unchanged"
+        ),
     )
 
 
@@ -469,14 +702,14 @@ def main() -> int:
             )
         policy = load_policy(Path(args.policy))
         run_independent_bundle_verifier(args)
-        validate_isolation_receipt(args, policy)
+        isolation = validate_isolation_receipt(args, policy)
         assets = load_portable_assets(Path(args.bundle_dir), policy)
         if args.check:
             result = run_check(args, policy, assets)
         elif args.apply:
             result = run_apply(args, policy, assets)
         else:
-            result = run_cleanup_verifier(args, policy)
+            result = run_cleanup_verifier(args, policy, isolation)
         if args.receipt_file:
             write_receipt(Path(args.receipt_file), result)
         emit_result(result)

scripts/backup/tests/test_signoz_metadata_export_contract.py

@@ -60,7 +60,10 @@ class MetadataHandler(BaseHTTPRequestHandler):
         return
 
     def _authorized(self) -> bool:
-        return self.headers.get("SIGNOZ-API-KEY") == TOKEN
+        return (
+            self.headers.get("SIGNOZ-API-KEY") == TOKEN
+            or self.headers.get("Authorization") == f"Bearer {TOKEN}"
+        )
 
     def _send_json(self, status: int, value: Any) -> None:
         payload = json.dumps(value, separators=(",", ":")).encode("utf-8")
@@ -219,20 +222,46 @@ def create_bundle(tmp_path: Path) -> tuple[Path, Path]:
 
 def isolation_receipt(tmp_path: Path, target_base_url: str) -> Path:
     policy = json.loads(POLICY.read_text(encoding="utf-8"))
+    identity = hashlib.sha256(
+        TRACE_ID.encode("utf-8")
+        + b"\0"
+        + RUN_ID.encode("utf-8")
+        + b"\0"
+        + WORK_ITEM_ID.encode("utf-8")
+    ).hexdigest()
+    short_id = identity[:16]
+    prefix = f"awoooi-signoz-md-restore-{short_id}"
+    port_text = target_base_url.rsplit(":", 1)[-1]
+    port = int(port_text) if port_text.isdigit() else 443
+    production_before = {
+        entry["container"]: {
+            "id": hashlib.sha256(entry["container"].encode()).hexdigest(),
+            "image_id": entry["image_id"],
+            "image_ref": entry["image_ref"],
+            "running": True,
+            "started_at": "2026-07-22T00:00:00Z",
+            "restart_count": 0,
+        }
+        for entry in policy["restore_isolation"]["production_continuity"].values()
+    }
     value = {
         "schema": "awoooi_signoz_metadata_restore_isolation_v1",
         "trace_id": TRACE_ID,
         "run_id": RUN_ID,
         "work_item_id": WORK_ITEM_ID,
         "target": {
-            "canonical_id": f"ephemeral-signoz-metadata-restore-{RUN_ID}",
+            "identity": identity,
+            "prefix": prefix,
+            "canonical_id": f"ephemeral-signoz-metadata-restore-{short_id}",
             "base_url_sha256": hashlib.sha256(
                 target_base_url.encode("utf-8")
             ).hexdigest(),
             "image_ref": "signoz/signoz:v0.113.0",
-            "image_id": f"sha256:{'a' * 64}",
+            "image_id": policy["restore_isolation"]["required_image_id"],
+            "images": policy["restore_isolation"]["images"],
             "image_present_locally": True,
             "image_pull_performed": False,
+            "image_build_performed": False,
             "production_network_attached": False,
             "production_volume_attached": False,
             "ephemeral_volumes_only": True,
@@ -241,7 +270,28 @@ def isolation_receipt(tmp_path: Path, target_base_url: str) -> Path:
             "network_scope": "loopback_and_internal_only",
             "resource_label_key": policy["restore_isolation"]["resource_label_key"],
             "resource_label_value": RUN_ID,
+            "identity_label_key": "com.awoooi.restore.identity",
+            "identity_label_value": identity,
+            "resources": {
+                "network": f"{prefix}-net",
+                "containers": [
+                    f"{prefix}-server",
+                    f"{prefix}-clickhouse",
+                    f"{prefix}-zookeeper",
+                ],
+                "volumes": [
+                    f"{prefix}-zk-data",
+                    f"{prefix}-zk-log",
+                    f"{prefix}-ch-data",
+                    f"{prefix}-server-data",
+                ],
+                "workspace": f"/tmp/{prefix}-private",
+            },
+            "listener": {"host": "127.0.0.1", "port": port},
+            "authentication_mode": "isolated_session_bearer",
         },
+        "production_before": production_before,
+        "image_inventory_before": ["sha256:test|test:fixture"],
     }
     path = tmp_path / "isolation.json"
     path.write_text(json.dumps(value), encoding="utf-8")
@@ -266,6 +316,8 @@ def restore_command(
         str(bundle),
         "--target-base-url",
         target_base_url,
+        "--auth-mode",
+        "isolated_session_bearer",
         "--isolation-receipt",
         str(isolation),
         "--trace-id",
@@ -349,6 +401,23 @@ def test_policy_rejects_unhashable_empty_collection_sentinel_as_contract_error(
     assert "Traceback" not in result.stderr
 
 
+def test_policy_pins_reviewed_isolated_restore_runtime() -> None:
+    policy = json.loads(POLICY.read_text(encoding="utf-8"))
+    isolation = policy["restore_isolation"]
+    assert isolation["required_image_id"] == (
+        "sha256:c17991d10966aedcb0d4d83805a0baf5310f1553d90611629036fcfaee8d969b"
+    )
+    assert isolation["startup_contract"]["reviewed"] is True
+    assert isolation["startup_contract"]["server_entrypoint"] == [
+        "./signoz",
+        "server",
+    ]
+    assert isolation["startup_contract"]["server_environment"] == {
+        "SIGNOZ_TELEMETRYSTORE_CLICKHOUSE_DSN": "tcp://clickhouse:9000",
+        "SIGNOZ_SQLSTORE_SQLITE_PATH": "/var/lib/signoz/signoz.db",
+    }
+
+
 def test_check_mode_writes_nothing(tmp_path: Path) -> None:
     output = tmp_path / "must-not-exist"
     result = subprocess.run(
@@ -654,7 +723,19 @@ def test_restore_check_and_semantic_apply_are_bounded(tmp_path: Path) -> None:
         assert json.loads(apply_result.stdout)["terminal"] == (
             "partial_degraded_cleanup_pending"
         )
-        phase_receipts = json.loads(apply_result.stdout)["phase_receipts"]
+        apply_receipt = json.loads(apply_result.stdout)
+        assert apply_receipt["restored_asset_counts"] == {
+            "dashboards": 1,
+            "alert_rules": 1,
+            "explorer_views_logs": 1,
+            "explorer_views_meter": 0,
+            "explorer_views_metrics": 0,
+            "explorer_views_traces": 0,
+        }
+        assert len(apply_receipt["portable_semantic_sha256"]) == 64
+        assert apply_receipt["completion_scope"] == "portable_assets_only"
+        assert apply_receipt["full_sqlite_completion_claim"] is False
+        phase_receipts = apply_receipt["phase_receipts"]
         assert [item["phase"] for item in phase_receipts] == [
             "sensor_source",
             "normalized_asset_identity",
@@ -693,12 +774,74 @@ def test_restore_rejects_non_loopback_target_before_writes(tmp_path: Path) -> No
     assert "restore_target_must_be_loopback" in result.stderr
 
 
+def test_restore_rejects_non_exact_isolation_image_before_writes(
+    tmp_path: Path,
+) -> None:
+    bundle, credential = create_bundle(tmp_path)
+    with socket.socket() as sock:
+        sock.bind(("127.0.0.1", 0))
+        port = sock.getsockname()[1]
+    target = f"http://127.0.0.1:{port}"
+    isolation = isolation_receipt(tmp_path, target)
+    value = json.loads(isolation.read_text(encoding="utf-8"))
+    value["target"]["image_id"] = f"sha256:{'a' * 64}"
+    isolation.write_text(json.dumps(value), encoding="utf-8")
+    isolation.chmod(0o600)
+
+    result = subprocess.run(
+        restore_command(
+            mode="--apply",
+            bundle=bundle,
+            target_base_url=target,
+            credential=credential,
+            isolation=isolation,
+        ),
+        text=True,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+        timeout=20,
+        check=False,
+    )
+
+    assert result.returncode == 1
+    assert "isolation_target_image_id_mismatch" in result.stderr
+
+
 def test_cleanup_verifier_requires_zero_labeled_residue(tmp_path: Path) -> None:
     bundle, _ = create_bundle(tmp_path)
     fake_bin = tmp_path / "bin"
     fake_bin.mkdir()
     docker = fake_bin / "docker"
-    docker.write_text("#!/bin/sh\nexit 0\n", encoding="utf-8")
+    policy = json.loads(POLICY.read_text(encoding="utf-8"))
+    production = {
+        entry["container"]: {
+            "Id": hashlib.sha256(entry["container"].encode()).hexdigest(),
+            "Image": entry["image_id"],
+            "Config": {"Image": entry["image_ref"]},
+            "State": {
+                "Running": True,
+                "StartedAt": "2026-07-22T00:00:00Z",
+            },
+            "RestartCount": 0,
+        }
+        for entry in policy["restore_isolation"]["production_continuity"].values()
+    }
+    docker.write_text(
+        "#!/usr/bin/env python3\n"
+        "import json, sys\n"
+        f"production = {production!r}\n"
+        "args = sys.argv[1:]\n"
+        "if args[:2] == ['container', 'inspect'] and args[2] in production:\n"
+        "    print(json.dumps([production[args[2]]]))\n"
+        "    raise SystemExit(0)\n"
+        "if args[:3] == ['image', 'ls', '--no-trunc']:\n"
+        "    print('sha256:test|test:fixture')\n"
+        "    raise SystemExit(0)\n"
+        "if len(args) >= 2 and args[1] == 'inspect':\n"
+        "    raise SystemExit(1)\n"
+        "raise SystemExit(0)\n",
+        encoding="utf-8",
+    )
     docker.chmod(0o755)
 
     with socket.socket() as sock:

scripts/backup/tests/test_signoz_metadata_isolated_restore.py

@@ -0,0 +1,597 @@
+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]
+            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"}},
+                "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"))
+    assert controller.validate_restore_policy(policy)["startup_contract"][
+        "reviewed"
+    ] is True
+
+    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)
+
+
+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)
+    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
+
+
+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": 3,
+            },
+        },
+    )
+    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 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
+
+
+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,
+    ) -> 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_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