Data Migrations During Hot Reload

When you update a content pack, component schemas may change. Data migrations ensure existing entities retain their data in a compatible format.

Understanding Migrations

When Migrations Are Needed

Migrations are required when you:

• Add fields to a component with no default value
• Remove fields that contained important data
• Rename fields in a component
• Change field types (e.g., int to float)
• Restructure data within a component

Migration Flow

Old Component Data -> Migration Function -> New Component Data
        v1.0.0              transform              v2.0.0

Migrations run automatically during hot reload after the new pack loads.

Creating Migrations

ComponentMigration

Define a migration with ComponentMigration:

from maid_engine.plugins.migration import ComponentMigration

def migrate_health_v1_to_v2(data: dict[str, Any]) -> dict[str, Any]:
    """Migrate HealthComponent from v1 to v2.

    Changes:
    - Renamed 'hp' to 'current'
    - Added 'maximum' field (defaults to current value)
    """
    return {
        "current": data.get("hp", 100),
        "maximum": data.get("hp", 100),
        "regeneration_rate": data.get("regeneration_rate", 0.0),
    }

migration = ComponentMigration(
    component_type="my_pack.components.HealthComponent",
    from_version="1.0.0",
    to_version="2.0.0",
    migrate_fn=migrate_health_v1_to_v2,
    description="Rename hp to current, add maximum field",
)

Registering Migrations

Register migrations with the HotReloadManager:

# In your content pack's on_load hook
async def on_load(self, engine: GameEngine) -> None:
    # Register migrations
    engine.hot_reload.register_migration(self.manifest.name, migration)

Or register directly:

engine.hot_reload.migration_registry.register_migration(migration)

Migration Registry

The MigrationRegistry manages all migrations and supports automatic path finding.

Automatic Chaining

If you have migrations for v1 -> v2 and v2 -> v3, the registry automatically chains them:

from maid_engine.plugins.migration import MigrationRegistry

registry = MigrationRegistry()

# Register individual migrations
registry.register_migration(migration_v1_v2)
registry.register_migration(migration_v2_v3)

# Get migration path from v1 to v3 (automatically chains)
migrations = registry.get_migrations(
    "my_pack.components.HealthComponent",
    from_ver="1.0.0",
    to_ver="3.0.0"
)

# Apply migrations in sequence
data = {"hp": 100}
for migration in migrations:
    data = migration.apply(data)

Checking Migration Paths

# Check if a path exists
has_path = registry.has_migration_path(
    "my_pack.components.HealthComponent",
    from_ver="1.0.0",
    to_ver="3.0.0"
)

# List all migrations for a component
all_migrations = registry.get_all_migrations("my_pack.components.HealthComponent")

# List all component types with migrations
component_types = registry.get_component_types()

Handling Orphaned Components

When a component type is removed or no migration path exists, the component becomes "orphaned."

Orphaned Component Strategies

Configure how orphaned components are handled:

from maid_engine.plugins.migration import (
    OrphanedComponentStrategy,
    OrphanedComponentHandler,
)

handler = OrphanedComponentHandler(
    strategy=OrphanedComponentStrategy.ARCHIVE
)

engine.hot_reload.orphaned_handler.strategy = OrphanedComponentStrategy.ARCHIVE

Strategy	Behavior
DISCARD	Log warning and remove the component (default)
KEEP	Log warning and keep data unchanged
ARCHIVE	Log warning and save to archive for later recovery
ERROR	Raise an error and abort the migration

Recovering Archived Components

# Get archived orphaned components
handler = engine.hot_reload.orphaned_handler
archived = handler.archive

for info in archived:
    print(f"Entity {info.entity_id}: {info.component_type}")
    print(f"  Reason: {info.reason}")
    print(f"  Data: {info.component_data}")

# Clear archive after processing
handler.clear_archive()

StatefulSystem Protocol

For systems that maintain internal state beyond components, implement StatefulSystem:

from typing import Any, Protocol, runtime_checkable
from maid_engine.core.ecs import System

@runtime_checkable
class StatefulSystem(Protocol):
    def capture_state(self) -> dict[str, Any]: ...
    def restore_state(self, state: dict[str, Any]) -> None: ...

Example Implementation

from uuid import UUID
from dataclasses import dataclass
from maid_engine.core.ecs import System
from maid_engine.core.world import World

@dataclass
class QueuedAction:
    entity_id: UUID
    action_type: str
    delay: float

    def to_dict(self) -> dict:
        return {
            "entity_id": str(self.entity_id),
            "action_type": self.action_type,
            "delay": self.delay,
        }

    @classmethod
    def from_dict(cls, data: dict) -> "QueuedAction":
        return cls(
            entity_id=UUID(data["entity_id"]),
            action_type=data["action_type"],
            delay=data["delay"],
        )


class ActionQueueSystem(System):
    """System that queues delayed actions."""

    priority = 10

    def __init__(self, world: World) -> None:
        super().__init__(world)
        self._queue: list[QueuedAction] = []
        self._processed_count: int = 0

    def capture_state(self) -> dict[str, Any]:
        """Capture state before hot reload."""
        return {
            "queue": [action.to_dict() for action in self._queue],
            "processed_count": self._processed_count,
        }

    def restore_state(self, state: dict[str, Any]) -> None:
        """Restore state after hot reload."""
        self._queue = [
            QueuedAction.from_dict(data)
            for data in state.get("queue", [])
        ]
        self._processed_count = state.get("processed_count", 0)

    async def update(self, delta: float) -> None:
        # Process queue
        for action in list(self._queue):
            action.delay -= delta
            if action.delay <= 0:
                await self._execute_action(action)
                self._queue.remove(action)
                self._processed_count += 1

    async def _execute_action(self, action: QueuedAction) -> None:
        # Execute the action
        pass

Using State Preservation

When reloading packs with stateful systems:

# Use reload_pack_with_state for automatic state preservation
result = await engine.hot_reload.reload_pack_with_state(
    "my-pack",
    new_pack=MyContentPack()
)

# State is automatically captured and restored

Migration Best Practices

1. Always Provide Defaults

def migrate_v1_to_v2(data: dict[str, Any]) -> dict[str, Any]:
    return {
        "name": data.get("name", "Unknown"),  # Default if missing
        "level": data.get("level", 1),
        "new_field": data.get("new_field", "default_value"),
    }

2. Handle Missing Data Gracefully

def migrate_v1_to_v2(data: dict[str, Any]) -> dict[str, Any]:
    # Handle completely empty data
    if not data:
        return {"current": 100, "maximum": 100}

    # Handle partial data
    current = data.get("hp") or data.get("current") or 100
    return {
        "current": current,
        "maximum": data.get("maximum", current),
    }

3. Keep Migrations Reversible When Possible

# Forward migration: v1 -> v2
def migrate_v1_to_v2(data: dict[str, Any]) -> dict[str, Any]:
    return {
        "value": data.get("old_value", 0),
        "_migrated_from": "v1",  # Track migration source
    }

# Reverse migration: v2 -> v1 (for rollback scenarios)
def migrate_v2_to_v1(data: dict[str, Any]) -> dict[str, Any]:
    return {
        "old_value": data.get("value", 0),
    }

4. Test Migrations Thoroughly

import pytest
from my_pack.migrations import migration_v1_to_v2

class TestMigrations:
    def test_migrate_v1_to_v2_full_data(self):
        old_data = {"hp": 50, "regeneration_rate": 1.0}
        new_data = migration_v1_to_v2.apply(old_data)

        assert new_data["current"] == 50
        assert new_data["maximum"] == 50
        assert new_data["regeneration_rate"] == 1.0

    def test_migrate_v1_to_v2_empty_data(self):
        old_data = {}
        new_data = migration_v1_to_v2.apply(old_data)

        assert new_data["current"] == 100
        assert new_data["maximum"] == 100

    def test_migrate_v1_to_v2_partial_data(self):
        old_data = {"hp": 75}
        new_data = migration_v1_to_v2.apply(old_data)

        assert new_data["current"] == 75
        assert new_data["maximum"] == 75

5. Document Migration Changes

migration = ComponentMigration(
    component_type="my_pack.components.InventoryComponent",
    from_version="1.0.0",
    to_version="2.0.0",
    migrate_fn=migrate_inventory_v1_to_v2,
    description="""
    Inventory v1 -> v2 migration:
    - Changed 'items' from list to dict keyed by slot
    - Added 'max_slots' field (default: 20)
    - Removed deprecated 'weight_limit' field
    """,
)

Migration Results

After hot reload, check migration results:

result = await engine.hot_reload.reload_pack("my-pack", new_pack)

print(f"Migrations run: {result.migrations_run}")

# Check reload history for detailed results
history = engine.hot_reload.get_reload_history()
for r in history:
    print(f"{r.pack_name}: {r.migrations_run} migrations, {r.elapsed_time:.3f}s")

Debugging Migrations

Enable Migration Logging

import logging

logging.getLogger("maid_engine.plugins.migration").setLevel(logging.DEBUG)
logging.getLogger("maid_engine.plugins.hot_reload").setLevel(logging.DEBUG)

Inspect Migration Registry

registry = engine.hot_reload.migration_registry

# List all registered migrations
for migration in registry.get_all_migrations():
    print(f"{migration.component_type}: {migration.from_version} -> {migration.to_version}")
    print(f"  {migration.description}")

# Check if path exists
has_path = registry.has_migration_path(
    "my_pack.components.HealthComponent",
    "1.0.0",
    "3.0.0"
)
print(f"Migration path exists: {has_path}")

Next Steps

• Hot Reload Overview - Review the hot reload architecture
• Development Workflow - Using hot reload during development