refactor(supervisor): rename vm_hash->vm_id, numeric vm_id->vm_index

Behaviour-neutral identifier rename so the supervisor objects say what they
mean (design: docs/plans/2026-06-19-supervisor-vm-id-rename-design.md).

Two axes:
- numeric local VM id 'vm_id: int' -> 'vm_index' (MicroVM, VmExecution.vm_index
  property + create(), network/, hypervisors/, controllers, get_unique_vm_index).
- string identity 'vm_hash' -> 'vm_id' on the supervisor objects (VmExecution,
  VmPool keys/params, LocalSupervisor) and agent reads of that attribute
  (run/custom_logs/tasks/migration).

Left unchanged: the controller's own 'vm_hash: ItemHash' param, the serialized
Configuration.vm_id(int)/vm_hash(str) fields and the '"vm_id"'/'"vm_hash"'
recreate-dict string keys, the wire CreateVmSpec.vm_id/VmInfo.vm_id, and genuine
ItemHash locals.

Verified: mypy unchanged at baseline (43/13); full supervisor+migration+network
suite green (3 env-only test_interfaces pyroute2/root failures excepted);
import-linter 4 kept/0 broken; grep sweep clean (no execution.vm_hash).

Author: odesenfans

src/aleph/vm/agent/custom_logs.py

@@ -29,7 +29,7 @@ def filter(self, record):
         if not vm_hash:
             vm_execution: VmExecution | None = ctx_current_execution.get(None)
             if vm_execution:
-                vm_hash = vm_execution.vm_hash
+                vm_hash = vm_execution.vm_id
 
         if not vm_hash:
             return False

src/aleph/vm/agent/tasks.py

@@ -257,7 +257,7 @@ async def _handle_domains_aggregate(
     has_local_instance = any(
         execution.is_instance
         and execution.vm
-        and (record := registry.get(execution.vm_hash)) is not None
+        and (record := registry.get(execution.vm_id)) is not None
         and record.message.address == address
         for execution in pool.executions.values()
     )

src/aleph/vm/hypervisors/firecracker/microvm.py

@@ -78,7 +78,7 @@ def supports_ipv6(self) -> bool:
 
 
 class MicroVM:
-    vm_id: int
+    vm_index: int
     use_jailer: bool
     firecracker_bin_path: Path
     jailer_bin_path: Path | None
@@ -96,15 +96,15 @@ class MicroVM:
     journal_stderr: BinaryIO | int | None = None
 
     def __repr__(self):
-        return f"<MicroVM {self.vm_id}>"
+        return f"<MicroVM {self.vm_index}>"
 
     def __str__(self):
-        return f"vm-{self.vm_id}"
+        return f"vm-{self.vm_index}"
 
     @property
     def namespace_path(self) -> str:
         firecracker_bin_name = os.path.basename(self.firecracker_bin_path)
-        return str(self.jailer_base_directory / firecracker_bin_name / str(self.vm_id))
+        return str(self.jailer_base_directory / firecracker_bin_name / str(self.vm_index))
 
     @property
     def jailer_path(self) -> str:
@@ -115,7 +115,7 @@ def socket_path(self) -> str:
         if self.use_jailer:
             return f"{self.jailer_path}/run/firecracker.socket"
         else:
-            return f"/tmp/firecracker-{self.vm_id}.socket"
+            return f"/tmp/firecracker-{self.vm_index}.socket"
 
     @property
     def vsock_path(self) -> str:
@@ -126,7 +126,7 @@ def vsock_path(self) -> str:
 
     def __init__(
         self,
-        vm_id: int,
+        vm_index: int,
         vm_hash: ItemHash,
         firecracker_bin_path: Path,
         jailer_base_directory: Path,
@@ -135,7 +135,7 @@ def __init__(
         init_timeout: float = 5.0,
         enable_log: bool = True,
     ):
-        self.vm_id = vm_id
+        self.vm_index = vm_index
         self.vm_hash = vm_hash
         self.use_jailer = use_jailer
         self.jailer_base_directory = jailer_base_directory
@@ -263,7 +263,7 @@ async def start_jailed_firecracker(self, config_path: Path) -> asyncio.subproces
         options = (
             str(self.jailer_bin_path),
             "--id",
-            str(self.vm_id),
+            str(self.vm_index),
             "--exec-file",
             str(self.firecracker_bin_path),
             "--uid",
@@ -433,15 +433,15 @@ async def unix_client_connected(reader: asyncio.StreamReader, _writer: asyncio.S
             raise MicroVMFailedInitError() from error
 
     async def shutdown(self) -> None:
-        logger.debug(f"Shutdown vm={self.vm_id}")
+        logger.debug(f"Shutdown vm={self.vm_index}")
         try:
             reader, writer = await asyncio.open_unix_connection(path=self.vsock_path)
         except (
             FileNotFoundError,
             ConnectionResetError,
             ConnectionRefusedError,
         ) as error:
-            logger.warning(f"VM={self.vm_id} cannot receive shutdown signal: {error.args}")
+            logger.warning(f"VM={self.vm_index} cannot receive shutdown signal: {error.args}")
             return
 
         try:
@@ -462,7 +462,7 @@ async def shutdown(self) -> None:
             if msg2 != b"STOPZ\n":
                 logger.warning(f"Unexpected response from VM: {msg2[:20]!r}")
         except ConnectionResetError as error:
-            logger.warning(f"ConnectionResetError in shutdown of {self.vm_id}: {error.args}")
+            logger.warning(f"ConnectionResetError in shutdown of {self.vm_index}: {error.args}")
 
     async def stop(self):
         if self.proc:
@@ -481,7 +481,7 @@ async def teardown(self):
         try:
             await asyncio.wait_for(self.shutdown(), timeout=5)
         except asyncio.TimeoutError:
-            logger.exception(f"Timeout during VM shutdown vm={self.vm_id}")
+            logger.exception(f"Timeout during VM shutdown vm={self.vm_index}")
         logger.debug("Waiting for one second for the process to shutdown")
         await asyncio.sleep(1)
         await self.stop()

src/aleph/vm/migration/helpers.py

@@ -30,16 +30,16 @@ async def graceful_shutdown(execution: VmExecution, timeout: int = GRACEFUL_SHUT
         client.system_powerdown()
         client.close()
     except Exception as e:
-        logger.warning("Failed to send system_powerdown for %s: %s", execution.vm_hash, e)
+        logger.warning("Failed to send system_powerdown for %s: %s", execution.vm_id, e)
 
     start = time.monotonic()
     while time.monotonic() - start < timeout:
         if execution.systemd_manager and not execution.systemd_manager.is_service_active(execution.controller_service):
-            logger.info("VM %s shut down gracefully", execution.vm_hash)
+            logger.info("VM %s shut down gracefully", execution.vm_id)
             return
         await asyncio.sleep(1)
 
-    logger.warning("VM %s did not shut down within %ds, forcing stop", execution.vm_hash, timeout)
+    logger.warning("VM %s did not shut down within %ds, forcing stop", execution.vm_id, timeout)
     if execution.systemd_manager:
         execution.systemd_manager.stop_and_disable(execution.controller_service)
 

src/aleph/vm/models.py

@@ -95,7 +95,7 @@ class VmExecution:
     """
 
     uuid: uuid.UUID  # Unique identifier of this execution
-    vm_hash: VmId
+    vm_id: VmId
     # The message-free description this execution is built from.
     spec: CreateVmSpec
     resources: (
@@ -144,7 +144,7 @@ async def update_port_redirects(self, requested_ports: dict[int, dict[str, bool]
 
             for protocol in SUPPORTED_PROTOCOL_FOR_REDIRECT:
                 if target[protocol]:
-                    add_port_redirect_rule(self.vm.vm_id, interface, host_port, vm_port, protocol)
+                    add_port_redirect_rule(self.vm.vm_index, interface, host_port, vm_port, protocol)
             self.mapped_ports[int(vm_port)] = {"host": host_port, **target}
             changed = True
 
@@ -155,7 +155,7 @@ async def update_port_redirects(self, requested_ports: dict[int, dict[str, bool]
             for protocol in SUPPORTED_PROTOCOL_FOR_REDIRECT:
                 if current[protocol] != target[protocol]:
                     if target[protocol]:
-                        add_port_redirect_rule(self.vm.vm_id, interface, host_port, vm_port, protocol)
+                        add_port_redirect_rule(self.vm.vm_index, interface, host_port, vm_port, protocol)
                     else:
                         remove_port_redirect_rule(interface, host_port, vm_port, protocol)
                     changed = True
@@ -164,7 +164,7 @@ async def update_port_redirects(self, requested_ports: dict[int, dict[str, bool]
 
         # Persist port mappings to dedicated table if anything changed
         if changed:
-            await save_port_mappings(self.vm_hash, self.mapped_ports)
+            await save_port_mappings(self.vm_id, self.mapped_ports)
 
     async def recreate_port_redirect_rules(self) -> None:
         """Recreate nftables port redirect rules from saved mapped_ports after restart.
@@ -215,15 +215,15 @@ async def recreate_port_redirect_rules(self) -> None:
                     host_port,
                     new_host_port,
                     vm_port,
-                    self.vm_hash,
+                    self.vm_id,
                 )
                 host_port = new_host_port
                 mapping["host"] = new_host_port
                 port_changed = True
 
             for protocol in protocols_to_create:
                 all_entities += build_port_redirect_entities(
-                    self.vm.vm_id,
+                    self.vm.vm_index,
                     interface,
                     host_port,
                     vm_port,
@@ -242,11 +242,11 @@ async def recreate_port_redirect_rules(self) -> None:
                     protocol,
                     host_port,
                     vm_port,
-                    self.vm_hash,
+                    self.vm_id,
                 )
 
         if port_changed:
-            await save_port_mappings(self.vm_hash, self.mapped_ports)
+            await save_port_mappings(self.vm_id, self.mapped_ports)
 
     async def removed_all_ports_redirection(self):
         if not self.vm:
@@ -323,12 +323,12 @@ def becomes_ready(self) -> Callable[[], Coroutine]:
         return self.ready_event.wait
 
     @property
-    def vm_id(self) -> int | None:
-        return self.vm.vm_id if self.vm else None
+    def vm_index(self) -> int | None:
+        return self.vm.vm_index if self.vm else None
 
     @property
     def controller_service(self) -> str:
-        return f"aleph-vm-controller@{self.vm_hash}.service"
+        return f"aleph-vm-controller@{self.vm_id}.service"
 
     @property
     def allocated_memory_mib(self) -> int:
@@ -349,19 +349,19 @@ def has_resources(self) -> bool:
             return True
 
     def __repr__(self):
-        return f"<VMExecution {type(self.vm).__name__} {self.vm_hash} {self.times.started_at}>"
+        return f"<VMExecution {type(self.vm).__name__} {self.vm_id} {self.times.started_at}>"
 
     def __init__(
         self,
-        vm_hash: VmId,
+        vm_id: VmId,
         vm_spec: CreateVmSpec,
         snapshot_manager: SnapshotManager | None = None,
         systemd_manager: SystemDManager | None = None,
         persistent: bool = False,
     ):
         self.init_task = None
         self.uuid = uuid.uuid1()  # uuid1() includes the hardware address and timestamp
-        self.vm_hash = vm_hash
+        self.vm_id = vm_id
         self.spec = vm_spec
         self.times = VmExecutionTimes(defined_at=datetime.now(tz=timezone.utc))
         self.ready_event = asyncio.Event()
@@ -390,7 +390,7 @@ def from_spec(
         The supervisor's machinery (prepare/create/start) reads only the spec.
         """
         return cls(
-            vm_hash=spec.vm_id,
+            vm_id=spec.vm_id,
             vm_spec=spec,
             snapshot_manager=snapshot_manager,
             systemd_manager=systemd_manager,
@@ -416,9 +416,9 @@ async def prepare(self) -> None:
             if self.spec.backend is Backend.FIRECRACKER:
                 self.resources = SpecProgramResources.from_spec(self.spec)
             elif self.spec.tee is not None:
-                self.resources = AlephQemuConfidentialResources.from_spec(self.spec, namespace=str(self.vm_hash))
+                self.resources = AlephQemuConfidentialResources.from_spec(self.spec, namespace=str(self.vm_id))
             else:
-                self.resources = AlephQemuResources.from_spec(self.spec, namespace=str(self.vm_hash))
+                self.resources = AlephQemuResources.from_spec(self.spec, namespace=str(self.vm_id))
             self.times.prepared_at = datetime.now(tz=timezone.utc)
 
     def uses_gpu(self, pci_host: str) -> bool:
@@ -429,7 +429,7 @@ def uses_gpu(self, pci_host: str) -> bool:
         return False
 
     def create(
-        self, vm_id: int, tap_interface: TapInterface | None = None, prepare: bool = True
+        self, vm_index: int, tap_interface: TapInterface | None = None, prepare: bool = True
     ) -> AlephVmControllerInterface:
         if not self.resources:
             msg = "Execution resources must be configured first"
@@ -439,8 +439,8 @@ def create(
         if self.spec.backend is Backend.FIRECRACKER:
             assert isinstance(self.resources, SpecProgramResources)
             self.vm = vm = SpecFirecrackerProgram(
-                vm_id=vm_id,
-                vm_hash=self.vm_hash,
+                vm_index=vm_index,
+                vm_hash=self.vm_id,
                 spec=self.spec,
                 resources=self.resources,
                 tap_interface=tap_interface,
@@ -454,8 +454,8 @@ def create(
             # SAFETY-CRITICAL: never fall through to the plain AlephQemuInstance.
             assert isinstance(self.resources, AlephQemuConfidentialResources)
             self.vm = vm = AlephQemuConfidentialInstance(
-                vm_id=vm_id,
-                vm_hash=self.vm_hash,
+                vm_index=vm_index,
+                vm_hash=self.vm_id,
                 resources=self.resources,
                 enable_networking=self.spec.network.internet_access,
                 confidential_policy=int(self.spec.tee.policy, 0),
@@ -465,8 +465,8 @@ def create(
             return vm
         assert isinstance(self.resources, AlephQemuResources)
         self.vm = vm = AlephQemuInstance(
-            vm_id=vm_id,
-            vm_hash=self.vm_hash,
+            vm_index=vm_index,
+            vm_hash=self.vm_id,
             resources=self.resources,
             enable_networking=self.spec.network.internet_access,
             hardware_resources=hardware_resources,
@@ -639,7 +639,7 @@ async def stop(self) -> None:
         # Prevent concurrent calls to stop() using a Lock
         async with self.stop_pending_lock:
             if self.times.stopped_at is not None:
-                logger.debug(f"VM={self.vm.vm_id} already stopped")
+                logger.debug(f"VM={self.vm.vm_index} already stopped")
                 return
             if self.persistent and self.systemd_manager:
                 self.systemd_manager.stop_and_disable(self.controller_service)
@@ -655,7 +655,7 @@ async def stop(self) -> None:
             self.times.stopped_at = datetime.now(tz=timezone.utc)
 
             if self.vm.support_snapshot and self.snapshot_manager:
-                await self.snapshot_manager.stop_for(self.vm_hash)
+                await self.snapshot_manager.stop_for(self.vm_id)
             self.stop_event.set()
             logger.info("%s stopped", self)
 
@@ -695,6 +695,6 @@ async def record_usage(self):
         await delete_record(execution_uuid=str(self.uuid))
         # Non-persistent VMs won't restart, so clean up their port mappings
         if not self.persistent:
-            await delete_port_mappings(self.vm_hash)
+            await delete_port_mappings(self.vm_id)
         if settings.EXECUTION_LOG_ENABLED:
             await save_execution_data(execution_uuid=self.uuid, execution_data=self.to_json())