Use Windows Computer Use Sandbox

#!/usr/bin/env python3
# /// script
# requires-python = ">=3.10"
# dependencies = [
#     "e2b==2.45.1",
#     "httpx==0.28.1",
#     "python-dotenv==1.2.3",
# ]
# ///
"""Full lifecycle demo for E2B Windows Template/Sandbox/Snapshot (region-independent).

Region, endpoint, image, and VPC settings are all specified in the .env file next to
the script. To switch regions, edit only the configuration file, not the script.

How to run:
    uv run windows_snapshot_lifecycle.py

Workflow:
    1. Build a Template from the Windows image
    2. Create the source Sandbox from the template_id returned in step 1
    3. Create a Snapshot from the sandbox_id returned in step 2
    4. Restore a new Sandbox from the snapshot_id returned in step 3
    5. Print the noVNC URL of the restored Sandbox

All configuration is read from the .env file next to the script with override=True forced,
so that values in .env replace existing environment variables in the current process. The
restored Sandbox is kept by default so that noVNC can be accessed immediately. Set CLEANUP_RESTORED_SANDBOX=true to clean it up when the script ends.
"""

from __future__ import annotations
from e2b.exceptions import SandboxException
from e2b.connection_config import ConnectionConfig
from e2b.api.client_sync import get_api_client
from e2b.api.client.api.snapshots import get_snapshots
from e2b import ApiParams, BuildInfo, Sandbox, Template, default_build_logger
import httpx

import base64
import hashlib
import os
import socket
import struct
import time
from pathlib import Path
from urllib.parse import quote

from dotenv import load_dotenv


# =============================================================================
# Load configuration: use override=True so that .env replaces variables in the process
# =============================================================================

ENV_PATH = Path(os.getenv("ENV_FILE", Path(
    __file__).with_name(".env"))).resolve()
if not ENV_PATH.is_file():
    raise FileNotFoundError(f".env file not found: {ENV_PATH}")
load_dotenv(dotenv_path=ENV_PATH, override=True)


# =============================================================================
# Common configuration and utilities
# =============================================================================

SNAPSHOT_DISCOVERY_TIMEOUT_SECONDS = 90
WINDOWS_SERVICE_PORT = int(os.getenv("WINDOWS_SERVICE_PORT", "5000"))
NOVNC_PORT = int(os.getenv("NOVNC_PORT", "6080"))

WINDOWS_READY_COMMAND = "curl -fsS --connect-timeout 1 --max-time 3 http://localhost:5000/health"


def require_env(name: str) -> str:
    value = os.getenv(name)
    if not value:
        raise RuntimeError(
            f"missing required env var: {name} (configure it in {ENV_PATH})")
    return value


def env_bool(name: str, default: bool = False) -> bool:
    value = os.getenv(name)
    return (
        default
        if value is None
        else value.strip().lower() in {"1", "true", "yes", "on"}
    )


def milestone(started_at: float, message: str) -> None:
    print(message)


def get_api_params() -> ApiParams:
    """Build the control plane parameters shared by Template, Sandbox, and Snapshot."""
    headers = {
        "X-E2B-Template-Build-Mode": os.getenv("E2B_TEMPLATE_BUILD_MODE", "micro"),
        "X-E2B-Template-Alpha-OS-Type": os.getenv(
            "E2B_TEMPLATE_OS_TYPE", "windows-osworld"
        ),
        "X-E2B-Template-Alpha-Port": str(WINDOWS_SERVICE_PORT),
        "X-E2B-Template-Alpha-Disk-Size": os.getenv("E2B_TEMPLATE_DISK_SIZE", "61440"),
        "X-E2B-Template-Alpha-Ready-Command":  WINDOWS_READY_COMMAND,
    }
    optional_headers = {
        "X-E2B-Template-Source-VPC-ID": os.getenv("E2B_TEMPLATE_SOURCE_VPC_ID"),
        "X-E2B-Template-Source-VSwitch-IDs": os.getenv(
            "E2B_TEMPLATE_SOURCE_VSWITCH_IDS"
        ),
        "X-E2B-Template-Source-Security-Group-ID": os.getenv(
            "E2B_TEMPLATE_SOURCE_SECURITY_GROUP_ID"
        ),
    }
    headers.update(
        {key: value for key, value in optional_headers.items() if value})

    return ApiParams(
        request_timeout=int(os.getenv("E2B_REQUEST_TIMEOUT_SECONDS", "600")),
        headers=headers,
        api_key=require_env("E2B_API_KEY"),
        domain=require_env("E2B_DOMAIN"),
        api_url=require_env("E2B_API_URL"),
        validate_api_key=False,
    )


# =============================================================================
# Step 1: build a Template from the Windows image
# =============================================================================


def build_windows_template(started_at: float) -> BuildInfo:
    rootfs_image = require_env("ROOTFS_IMAGE")
    cpu_count = int(os.getenv("E2B_TEMPLATE_CPU_COUNT", "16"))
    memory_mb = int(os.getenv("E2B_TEMPLATE_MEMORY_MB", "16448"))
    os_type = os.getenv("E2B_TEMPLATE_OS_TYPE", "windows-osworld")
    name_prefix = os.getenv("E2B_TEMPLATE_NAME_PREFIX",
                            "windows-snapshot-lifecycle")
    template_name = f"{name_prefix}-{os_type}-{time.time_ns()}"

    milestone(
        started_at, f"Step 1/5 building Windows Template from {rootfs_image}")
    template = Template().build_in_background(
        template=Template()
        .from_image(image=rootfs_image),
        name=template_name,
        cpu_count=cpu_count,
        memory_mb=memory_mb,
        skip_cache=env_bool("E2B_TEMPLATE_SKIP_CACHE", False),
        on_build_logs=default_build_logger(),
        **get_api_params(),
    )
    milestone(
        started_at,
        f"Template build accepted: template_id={template.template_id}, build_id={template.build_id}",
    )

    deadline = time.monotonic() + int(
        os.getenv("TEMPLATE_BUILD_TIMEOUT_SECONDS", "3600")
    )
    interval = int(os.getenv("TEMPLATE_BUILD_POLL_INTERVAL_SECONDS", "10"))
    while time.monotonic() < deadline:
        status = Template.get_build_status(
            template, logs_offset=0, **get_api_params())
        state = status.status.value
        if state == "ready":
            milestone(
                started_at, f"Step 1/5 complete: template_id={template.template_id}"
            )
            return template
        if state == "error":
            raise RuntimeError(f"template build failed: {status}")
        if state not in {"building", "waiting"}:
            raise RuntimeError(f"unexpected template build status: {status}")
        print(
            f"template_id={template.template_id} status={state}; waiting {interval}s..."
        )
        time.sleep(interval)

    raise TimeoutError(
        f"template build timed out: template_id={template.template_id}")


# =============================================================================
# Step 2 / Step 4: create a Sandbox from a Template ID or a Snapshot ID
# =============================================================================


def find_recent_sandbox(source_id: str) -> Sandbox | None:
    """If the connection drops before the create response arrives, find the most recent instance created from the same Template/Snapshot."""
    try:
        paginator = Sandbox.list(limit=50, **get_api_params())
        candidates = []
        while paginator.has_next:
            for info in paginator.next_items():
                if info.template_id == source_id and str(info.state) not in {
                    "killed",
                    "dead",
                    "error",
                }:
                    candidates.append(info)
        if not candidates:
            return None
        latest = max(candidates, key=lambda info: info.started_at)
        return Sandbox.connect(
            latest.sandbox_id,
            timeout=int(os.getenv("SANDBOX_TIMEOUT_SECONDS", "900")),
            **get_api_params(),
        )
    except Exception as exc:
        print(f"sandbox recovery lookup failed: {type(exc).__name__}: {exc}")
        return None


def create_sandbox(source_id: str, started_at: float, step: int) -> Sandbox:
    """Create a Sandbox from template_id or snapshot_id, and recover if the connection drops during Windows sandbox creation."""
    timeout = int(os.getenv("SANDBOX_TIMEOUT_SECONDS", "900"))
    retry_waits = (30, 60, 90, 120)

    for attempt in range(1, len(retry_waits) + 2):
        try:
            sandbox = Sandbox.create(
                template=source_id,
                timeout=timeout,
                **get_api_params(),
            )
            milestone(
                started_at,
                f"Step {step}/5 complete: source_id={source_id}, sandbox_id={sandbox.sandbox_id}",
            )
            return sandbox
        except httpx.RemoteProtocolError as exc:
            if attempt > len(retry_waits):
                raise
            wait_seconds = retry_waits[attempt - 1]
            print(
                f"Sandbox.create response disconnected ({attempt}/5): {exc}; "
                f"waiting {wait_seconds}s before recovery"
            )
            time.sleep(wait_seconds)
            recovered = find_recent_sandbox(source_id)
            if recovered is not None:
                milestone(
                    started_at,
                    f"Step {step}/5 recovered: source_id={source_id}, "
                    f"sandbox_id={recovered.sandbox_id}",
                )
                return recovered

    raise RuntimeError(f"failed to create sandbox from source_id={source_id}")


def kill_sandbox(sandbox: Sandbox | None, label: str) -> None:
    if sandbox is None:
        return
    try:
        sandbox.kill()
        print(f"cleaned {label}: sandbox_id={sandbox.sandbox_id}")
    except Exception as exc:
        print(
            f"warning: failed to clean {label} sandbox_id={sandbox.sandbox_id}: "
            f"{type(exc).__name__}: {exc}"
        )


# =============================================================================
# Windows data plane: wait for the service and write or verify the snapshot marker
# =============================================================================


def execute_windows(sandbox: Sandbox, command: list[str]) -> dict:
    access_token = sandbox._envd_access_token
    if not access_token:
        raise RuntimeError(
            f"sandbox has no access token: sandbox_id={sandbox.sandbox_id}"
        )

    response = httpx.post(
        f"http://{sandbox.get_host(WINDOWS_SERVICE_PORT)}/execute",
        headers={"X-Access-Token": access_token},
        json={"command": command},
        timeout=30,
    )
    if response.is_error:
        raise RuntimeError(
            f"Windows /execute failed: status={response.status_code}, body={response.text}"
        )
    result = response.json()
    if result.get("returncode") != 0 or result.get("status") != "success":
        raise RuntimeError(
            f"Windows command failed: command={command}, result={result}"
        )
    return result


def wait_for_windows_service(
    sandbox: Sandbox, timeout: int = 180, interval: int = 5
) -> None:
    deadline = time.monotonic() + timeout
    last_error: Exception | None = None
    while time.monotonic() < deadline:
        try:
            execute_windows(sandbox, ["whoami"])
            return
        except Exception as exc:
            last_error = exc
            time.sleep(interval)
    raise TimeoutError(
        f"Windows service not ready: sandbox_id={sandbox.sandbox_id}, error={last_error}"
    )


# =============================================================================
# Step 3: create a Snapshot from the source sandbox_id
# =============================================================================


def find_snapshot_id(snapshots: list, snapshot_name: str) -> str | None:
    expected_suffix = f"/{snapshot_name}:default"
    for snapshot in snapshots:
        if any(
            name == snapshot_name or name.endswith(expected_suffix)
            for name in snapshot.names
        ):
            return snapshot.snapshot_id
    return None


def lookup_snapshot_id(snapshot_name: str, sandbox_id: str) -> str | None:
    client = get_api_client(ConnectionConfig(**get_api_params()))
    response = get_snapshots.sync_detailed(
        client=client,
        sandbox_id=sandbox_id,
        name=snapshot_name,
        limit=100,
    )
    if response.status_code == 200 and isinstance(response.parsed, list):
        return find_snapshot_id(response.parsed, snapshot_name)
    return None


def discover_snapshot_after_disconnect(snapshot_name: str, sandbox_id: str) -> str:
    """If the connection drops before the create_snapshot response arrives, query the server-side result instead of replaying the non-idempotent request."""
    deadline = time.monotonic() + SNAPSHOT_DISCOVERY_TIMEOUT_SECONDS
    while time.monotonic() < deadline:
        snapshot_id = lookup_snapshot_id(snapshot_name, sandbox_id)
        if snapshot_id:
            return snapshot_id
        time.sleep(2)
    raise TimeoutError(
        f"snapshot not discoverable after disconnected response: "
        f"sandbox_id={sandbox_id}, name={snapshot_name}"
    )


def create_snapshot(sandbox: Sandbox, started_at: float) -> str:
    wait_for_windows_service(sandbox)
    probe_novnc_first_frame(
        sandbox,
        started_at,
        timeout=int(os.getenv("NOVNC_READY_TIMEOUT_SECONDS", "600")),
    )
    snapshot_name = f"windows-snapshot-{time.time_ns()}"
    milestone(
        started_at,
        f"Step 3/5 creating Snapshot from sandbox_id={sandbox.sandbox_id}",
    )

    max_attempts = int(os.getenv("SNAPSHOT_CREATE_MAX_ATTEMPTS", "12"))
    retry_interval = int(
        os.getenv("SNAPSHOT_CREATE_RETRY_INTERVAL_SECONDS", "10"))
    snapshot_id = ""
    for attempt in range(1, max_attempts + 1):
        try:
            snapshot_info = sandbox.create_snapshot(
                name=snapshot_name, **get_api_params()
            )
            snapshot_id = snapshot_info.snapshot_id
            if not snapshot_id:
                raise RuntimeError(
                    f"snapshot creation returned no ID: {snapshot_info}")
            break
        except httpx.RemoteProtocolError as exc:
            print(f"snapshot response disconnected; discovering result: {exc}")
            snapshot_id = discover_snapshot_after_disconnect(
                snapshot_name, sandbox.sandbox_id
            )
            break
        except SandboxException as exc:
            if "503" not in str(exc) or attempt >= max_attempts:
                raise
            snapshot_id = lookup_snapshot_id(
                snapshot_name, sandbox.sandbox_id) or ""
            if snapshot_id:
                break
            print(
                f"snapshot is temporarily unavailable ({attempt}/{max_attempts}); "
                f"retrying in {retry_interval}s"
            )
            time.sleep(retry_interval)

    milestone(
        started_at,
        f"Step 3/5 complete: sandbox_id={sandbox.sandbox_id}, snapshot_id={snapshot_id}",
    )
    return snapshot_id


# =============================================================================
# Step 5: print the noVNC URL of the Sandbox restored from snapshot_id
# =============================================================================


class WebSocketBinaryStream:
    """A minimal WebSocket binary stream for noVNC/RFB readiness probing."""

    def __init__(self, sock: socket.socket, initial_data: bytes = b"") -> None:
        self.sock = sock
        self.raw_buffer = bytearray(initial_data)
        self.data_buffer = bytearray()

    def _recv_raw_exact(self, size: int) -> bytes:
        while len(self.raw_buffer) < size:
            chunk = self.sock.recv(65536)
            if not chunk:
                raise ConnectionError("WebSocket closed by peer")
            self.raw_buffer.extend(chunk)
        result = bytes(self.raw_buffer[:size])
        del self.raw_buffer[:size]
        return result

    def send_binary(self, payload: bytes, opcode: int = 0x2) -> None:
        mask = os.urandom(4)
        length = len(payload)
        header = bytearray([0x80 | opcode])
        if length < 126:
            header.append(0x80 | length)
        elif length <= 0xFFFF:
            header.append(0x80 | 126)
            header.extend(struct.pack("!H", length))
        else:
            header.append(0x80 | 127)
            header.extend(struct.pack("!Q", length))
        masked = bytes(value ^ mask[index % 4]
                       for index, value in enumerate(payload))
        self.sock.sendall(bytes(header) + mask + masked)

    def _recv_data_frame(self) -> bytes:
        while True:
            first, second = self._recv_raw_exact(2)
            opcode = first & 0x0F
            masked = bool(second & 0x80)
            length = second & 0x7F
            if length == 126:
                length = struct.unpack("!H", self._recv_raw_exact(2))[0]
            elif length == 127:
                length = struct.unpack("!Q", self._recv_raw_exact(8))[0]
            mask = self._recv_raw_exact(4) if masked else b""
            payload = self._recv_raw_exact(length)
            if masked:
                payload = bytes(
                    value ^ mask[index % 4] for index, value in enumerate(payload)
                )
            if opcode == 0x8:
                raise ConnectionError("WebSocket close frame received")
            if opcode == 0x9:
                self.send_binary(payload, opcode=0xA)
                continue
            if opcode in {0x0, 0x2}:
                return payload

    def read_exact(self, size: int) -> bytes:
        while len(self.data_buffer) < size:
            self.data_buffer.extend(self._recv_data_frame())
        result = bytes(self.data_buffer[:size])
        del self.data_buffer[:size]
        return result

    def discard_exact(self, size: int) -> None:
        remaining = size
        while remaining:
            if self.data_buffer:
                consumed = min(remaining, len(self.data_buffer))
                del self.data_buffer[:consumed]
                remaining -= consumed
                continue
            self.data_buffer.extend(self._recv_data_frame())


def connect_websocket(
    host: str, access_token: str, timeout: float
) -> WebSocketBinaryStream:
    sock = socket.create_connection((host, 80), timeout=timeout)
    sock.settimeout(timeout)
    key = base64.b64encode(os.urandom(16)).decode("ascii")
    request = (
        f"GET /websockify?accessToken={access_token} HTTP/1.1\r\n"
        f"Host: {host}\r\n"
        "Upgrade: websocket\r\n"
        "Connection: Upgrade\r\n"
        f"Origin: http://{host}\r\n"
        f"Sec-WebSocket-Key: {key}\r\n"
        "Sec-WebSocket-Version: 13\r\n"
        "Sec-WebSocket-Protocol: binary\r\n\r\n"
    )
    sock.sendall(request.encode("ascii"))
    response = bytearray()
    while b"\r\n\r\n" not in response:
        chunk = sock.recv(4096)
        if not chunk:
            raise ConnectionError("connection closed during WebSocket upgrade")
        response.extend(chunk)
    headers, initial_data = bytes(response).split(b"\r\n\r\n", 1)
    status_line = headers.split(b"\r\n", 1)[0]
    if b" 101 " not in status_line:
        sock.close()
        raise ConnectionError(
            f"WebSocket upgrade failed: {status_line.decode(errors='replace')}"
        )
    expected_accept = base64.b64encode(
        hashlib.sha1(
            (key + "258EAFA5-E914-47DA-95CA-C5AB0DC85B11").encode()).digest()
    )
    if expected_accept.lower() not in headers.lower():
        sock.close()
        raise ConnectionError("invalid Sec-WebSocket-Accept response")
    return WebSocketBinaryStream(sock, initial_data)


def probe_novnc_first_frame(
    sandbox: Sandbox, started_at: float, timeout: int = 600
) -> None:
    """Complete the RFB handshake and read the first frame to confirm that the Windows desktop has started."""
    host = f"{NOVNC_PORT}-{sandbox.sandbox_id}.{require_env('E2B_DOMAIN')}"
    access_token = sandbox._envd_access_token
    if not access_token:
        raise RuntimeError(
            f"sandbox has no access token: sandbox_id={sandbox.sandbox_id}"
        )

    deadline = time.monotonic() + timeout
    last_error: Exception | None = None
    while time.monotonic() < deadline:
        stream: WebSocketBinaryStream | None = None
        try:
            stream = connect_websocket(
                host, access_token, min(5, deadline - time.monotonic())
            )
            version = stream.read_exact(12)
            if not version.startswith(b"RFB "):
                raise RuntimeError(f"unexpected RFB version: {version!r}")
            stream.send_binary(version)

            security_type_count = stream.read_exact(1)[0]
            if security_type_count == 0:
                reason_length = struct.unpack("!I", stream.read_exact(4))[0]
                reason = stream.read_exact(
                    reason_length).decode(errors="replace")
                raise RuntimeError(f"RFB server rejected connection: {reason}")
            security_types = list(stream.read_exact(security_type_count))
            if 1 not in security_types:
                raise RuntimeError(
                    f"unsupported RFB security types: {security_types}")
            stream.send_binary(b"\x01")
            security_result = struct.unpack("!I", stream.read_exact(4))[0]
            if security_result != 0:
                raise RuntimeError(
                    f"RFB security negotiation failed: {security_result}"
                )

            stream.send_binary(b"\x01")
            width, height = struct.unpack("!HH", stream.read_exact(4))
            pixel_format = stream.read_exact(16)
            bytes_per_pixel = pixel_format[0] // 8
            name_length = struct.unpack("!I", stream.read_exact(4))[0]
            desktop_name = stream.read_exact(
                name_length).decode(errors="replace")

            stream.send_binary(struct.pack("!BBHi", 2, 0, 1, 0))
            stream.send_binary(struct.pack(
                "!BBHHHH", 3, 0, 0, 0, width, height))
            while True:
                message_type = stream.read_exact(1)[0]
                if message_type == 0:
                    header = stream.read_exact(3)
                    rectangle_count = struct.unpack("!H", header[1:])[0]
                    if rectangle_count:
                        for _ in range(rectangle_count):
                            rectangle = stream.read_exact(12)
                            _, _, rect_width, rect_height, encoding = struct.unpack(
                                "!HHHHi", rectangle
                            )
                            if encoding != 0:
                                raise RuntimeError(
                                    f"unexpected framebuffer encoding: {encoding}"
                                )
                            stream.discard_exact(
                                rect_width * rect_height * bytes_per_pixel
                            )
                        milestone(
                            started_at,
                            f"noVNC first frame ready: {width}x{height} {desktop_name!r}",
                        )
                        return
                elif message_type == 1:
                    header = stream.read_exact(5)
                    color_count = struct.unpack("!H", header[3:])[0]
                    stream.read_exact(color_count * 6)
                elif message_type == 2:
                    continue
                elif message_type == 3:
                    header = stream.read_exact(5)
                    text_length = struct.unpack("!I", header[1:])[0]
                    stream.read_exact(text_length)
                else:
                    raise RuntimeError(
                        f"unexpected RFB server message type: {message_type}"
                    )
        except (OSError, ConnectionError, TimeoutError) as exc:
            last_error = exc
            time.sleep(1)
        finally:
            if stream is not None:
                stream.sock.close()

    raise TimeoutError(
        f"noVNC first frame not ready: sandbox_id={sandbox.sandbox_id}, "
        f"error={last_error}"
    )


def get_novnc_url(sandbox: Sandbox) -> str:
    access_token = sandbox._envd_access_token
    if not access_token:
        raise RuntimeError(
            f"sandbox has no access token: sandbox_id={sandbox.sandbox_id}"
        )
    websocket_path = quote(f"websockify?accessToken={access_token}", safe="")
    scheme = os.getenv("SANDBOX_PUBLIC_SCHEME", "http")
    return (
        f"{scheme}://{NOVNC_PORT}-{sandbox.sandbox_id}.{require_env('E2B_DOMAIN')}"
        f"/vnc.html#autoconnect=true&resize=scale&path={websocket_path}"
    )


def wait_for_novnc(sandbox: Sandbox, timeout: int = 600, interval: int = 3) -> str:
    novnc_url = get_novnc_url(sandbox)
    deadline = time.monotonic() + timeout
    last_error: Exception | None = None
    while time.monotonic() < deadline:
        try:
            response = httpx.get(novnc_url, timeout=5)
            if response.status_code == 200 and "noVNC" in response.text:
                return novnc_url
            last_error = RuntimeError(f"HTTP {response.status_code}")
        except (httpx.HTTPError, OSError) as exc:
            last_error = exc
        time.sleep(interval)
    raise TimeoutError(
        f"noVNC not ready: sandbox_id={sandbox.sandbox_id}, error={last_error}"
    )


# =============================================================================
# Main workflow: run the 5 steps in sequence and clean up resources as configured
# =============================================================================


def main() -> None:
    started_at = time.perf_counter()
    source_sandbox: Sandbox | None = None
    restored_sandbox: Sandbox | None = None
    workflow_succeeded = False

    template_id = ""
    snapshot_id = ""
    novnc_url = ""

    try:
        # Step 1: Windows image -> Template
        template = build_windows_template(started_at)
        template_id = template.template_id

        # Step 2: template_id -> source Sandbox
        source_sandbox = create_sandbox(template_id, started_at, step=2)

        # Step 3: source sandbox_id -> Snapshot
        snapshot_id = create_snapshot(source_sandbox, started_at)

        # Step 4: snapshot_id -> restored Sandbox
        restored_sandbox = create_sandbox(snapshot_id, started_at, step=4)
        wait_for_windows_service(restored_sandbox)
        identity = str(
            execute_windows(restored_sandbox, ["whoami"]).get("output", "")
        ).strip()
        milestone(
            started_at,
            f"Restored Sandbox is ready: snapshot_id={snapshot_id}, "
            f"restored_sandbox_id={restored_sandbox.sandbox_id}, whoami={identity!r}",
        )

        # Step 5: print the noVNC URL of the restored Sandbox
        probe_novnc_first_frame(
            restored_sandbox,
            started_at,
            timeout=int(os.getenv("NOVNC_READY_TIMEOUT_SECONDS", "600")),
        )
        novnc_url = wait_for_novnc(restored_sandbox)
        milestone(
            started_at,
            f"Step 5/5 complete: snapshot_id={snapshot_id}, noVNC is ready",
        )
        workflow_succeeded = True
    finally:
        if not env_bool("KEEP_SOURCE_SANDBOX", False):
            kill_sandbox(source_sandbox, "source")

        cleanup_restored = env_bool("CLEANUP_RESTORED_SANDBOX", False)
        if restored_sandbox is not None and (
            cleanup_restored or not workflow_succeeded
        ):
            kill_sandbox(restored_sandbox, "restored")

    if not workflow_succeeded or restored_sandbox is None:
        raise RuntimeError(
            "Windows Template/Sandbox/Snapshot lifecycle did not complete"
        )

    print("\nSUCCESS: Windows Template -> Sandbox -> Snapshot -> Restore -> noVNC\n")
    print(f"template_id={template_id}")
    print(f"snapshot_id={snapshot_id}")
    print(f"restored_sandbox_id={restored_sandbox.sandbox_id}")
    print(f"restored_sandbox_novnc_url={novnc_url}")
    if not env_bool("CLEANUP_RESTORED_SANDBOX", False):
        print(
            "restored sandbox is intentionally kept alive; "
            "set CLEANUP_RESTORED_SANDBOX=true to clean it up automatically"
        )


if __name__ == "__main__":
    main()