Versioning & Migration

Schema evolution, state migration, backward compatibility, and safe deployments.

Versioning Strategy

Semantic Versioning for State

State Version: MAJOR.MINOR.PATCH
├── MAJOR: Breaking changes (incompatible schema)
├── MINOR: Backward-compatible additions
└── PATCH: Bug fixes, internal refactoring

Version Locations

Backward Compatibility Rules

Never Break

1. Never remove fields — Mark deprecated, keep indefinitely
2. Never change field types — Add new field with new type
3. Never change field names — Add new field with new name
4. Never change enum values — Add new values only
5. Never change required → optional — Can break consumers

Safe Changes

Migration Patterns

1. Expand-Contract (Parallel Run)

Phase 1: Expand
  - Add new field (nullable)
  - Write to both old + new
  - Deploy

Phase 2: Migrate Data
  - Backfill new field from old
  - Verify

Phase 3: Contract
  - Read from new field
  - Stop writing old field
  - Deploy

Phase 4: Cleanup
  - Drop old field
  - Deploy

2. Dual-Write with Validation

class DualWriteMigrator:
    def __init__(self, old_store, new_store, validator):
        self.old = old_store
        self.new = new_store
        self.validator = validator

    async def write(self, key: str, value: dict):
        # Write to both
        await asyncio.gather(
            self.old.write(key, value),
            self.new.write(key, self.transform(value))
        )

    async def validate(self, sample_size=1000):
        # Compare random samples
        keys = await self.old.random_keys(sample_size)
        mismatches = []
        for key in keys:
            old = await self.old.read(key)
            new = await self.new.read(key)
            if not self.validator(old, new):
                mismatches.append((key, old, new))
        return mismatches

3. Shadow Reads

class ShadowReader:
    def __init__(self, primary, shadow, comparator):
        self.primary = primary
        self.shadow = shadow
        self.comparator = comparator

    async def read(self, key: str):
        # Read from primary (source of truth)
        primary_result = await self.primary.read(key)

        # Shadow read (async, non-blocking)
        asyncio.create_task(self._shadow_read(key))

        return primary_result

    async def _shadow_read(self, key: str):
        try:
            shadow_result = await self.shadow.read(key)
            primary_result = await self.primary.read(key)
            if not self.comparator(primary_result, shadow_result):
                await self.alert_mismatch(key, primary_result, shadow_result)
        except Exception as e:
            await self.alert_error(key, e)

Migration Tooling

Migration Framework

class Migration:
    def __init__(self, version: str, description: str):
        self.version = version
        self.description = description

    async def up(self, store) -> MigrationResult:
        raise NotImplementedError()

    async def down(self, store) -> MigrationResult:
        raise NotImplementedError()

    async def validate(self, store) -> ValidationResult:
        raise NotImplementedError()

class MigrationRunner:
    def __init__(self, migrations: list[Migration], store):
        self.migrations = sorted(migrations, key=lambda m: m.version)
        self.store = store

    async def run(self, target_version: str = None):
        applied = await self.get_applied_versions()

        for migration in self.migrations:
            if migration.version in applied:
                continue
            if target_version and migration.version > target_version:
                break

            print(f"Running migration {migration.version}: {migration.description}")
            result = await migration.up(self.store)

            if not result.success:
                raise MigrationFailed(migration.version, result.error)

            await self.record_applied(migration.version)
            print(f"✓ Migration {migration.version} completed")

    async def rollback(self, target_version: str):
        applied = await self.get_applied_versions()
        to_rollback = [m for m in reversed(self.migrations) 
                       if m.version in applied and m.version > target_version]

        for migration in to_rollback:
            print(f"Rolling back {migration.version}")
            result = await migration.down(self.store)
            if not result.success:
                raise RollbackFailed(migration.version, result.error)
            await self.record_rolled_back(migration.version)

State Migration Examples

Example 1: Add Field with Default

class AddCostTrackingMigration(Migration):
    version = "2.1.0"
    description = "Add cost tracking to agent state"

    async def up(self, store):
        # Add field with default
        cursor = store.db.execute("""
            ALTER TABLE agent_state ADD COLUMN cost_usd REAL DEFAULT 0.0
        """)

        # Backfill existing records
        store.db.execute("""
            UPDATE agent_state 
            SET cost_usd = tokens_used * 0.000002
            WHERE cost_usd = 0.0
        """)

    async def down(self, store):
        store.db.execute("ALTER TABLE agent_state DROP COLUMN cost_usd")

    async def validate(self, store):
        # Verify all records have cost_usd
        count = store.db.execute("SELECT COUNT(*) FROM agent_state WHERE cost_usd IS NULL").fetchone()[0]
        return ValidationResult(success=count == 0)

Example 2: Rename Field (Expand-Contract)

class RenameOutputToResultMigration(Migration):
    version = "3.0.0"
    description = "Rename tool_call.output to result"

    async def up(self, store):
        # Phase 1: Add new column
        store.db.execute("ALTER TABLE tool_calls ADD COLUMN result TEXT")

        # Phase 2: Backfill
        store.db.execute("UPDATE tool_calls SET result = output WHERE result IS NULL")

        # Phase 3: Verify
        null_count = store.db.execute("SELECT COUNT(*) FROM tool_calls WHERE result IS NULL").fetchone()[0]
        assert null_count == 0

        # Phase 4: Switch reads (code deploy needed between phases)
        # Phase 5: Drop old column (later migration)

    async def down(self, store):
        store.db.execute("UPDATE tool_calls SET output = result WHERE output IS NULL")
        store.db.execute("ALTER TABLE tool_calls DROP COLUMN result")

Example 3: Event Schema Migration

class EventSchemaMigration(Migration):
    version = "2.0.0"
    description = "Migrate ToolCallCompleted event from v1 to v2"

    async def up(self, store):
        # Add upcaster for event store
        store.register_upcaster("ToolCallCompleted", ToolCallUpcasterV1ToV2())

        # Optionally rewrite historical events (expensive!)
        # cursor = store.db.execute("""
        #     UPDATE events SET payload = jsonb_set(payload, '{result}', payload->'output'),
        #                        version = 2
        #     WHERE event_type = 'ToolCallCompleted' AND version = 1
        # """)

    async def down(self, store):
        store.unregister_upcaster("ToolCallCompleted")

Migration Safety

Pre-Migration Checklist

• [ ] Migration tested in staging with production data copy
• [ ] Rollback plan documented and tested
• [ ] Monitoring alerts configured for migration metrics
• [ ] Rollback time estimated (< 10 min for hotfix)
• [ ] Stakeholders notified of maintenance window
• [ ] Rollback tested in staging
• [ ] Data backup verified before migration

Canary Migration

class CanaryMigration:
    def __init__(self, migration, canary_percentage=5):
        self.migration = migration
        self.canary_pct = canary_percentage

    async def run(self, store):
        # 1. Get canary keys
        all_keys = await store.get_all_keys()
        canary_keys = random.sample(all_keys, int(len(all_keys) * self.canary_pct / 100))

        # 2. Run migration on canary
        for key in canary_keys:
            await self.migration.migrate_key(store, key)

        # 3. Validate canary
        await asyncio.sleep(300)  # Observe for 5 min
        validation = await self.migration.validate_canary(store, canary_keys)

        if not validation.success:
            await self.rollback_canary(store, canary_keys)
            raise CanaryFailed(validation.errors)

        # 4. Full rollout
        await self.migration.run_full(store)

Testing Migrations

Unit Tests

class TestMigrations:
    @pytest.fixture
    def store(self):
        return TestStore()

    @pytest.mark.parametrize("migration", ALL_MIGRATIONS)
    async def test_migration_up_down(self, migration, store):
        # Apply
        await migration.up(store)

        # Validate
        result = await migration.validate(store)
        assert result.success

        # Rollback
        await migration.down(store)

        # Verify rollback
        assert store.schema_version == PREVIOUS_VERSION

    async def test_migration_idempotent(self, migration, store):
        await migration.up(store)
        await migration.up(store)  # Should not fail
        result = await migration.validate(store)
        assert result.success

Integration Tests

async def test_full_migration_path():
    store = ProductionLikeStore()

    # Apply all migrations in order
    for migration in ALL_MIGRATIONS:
        await migration.up(store)
        assert await migration.validate(store)

    # Verify final state
    final_state = store.get_state()
    assert final_state.schema_version == LATEST_VERSION

    # Test rollback from latest
    for migration in reversed(ALL_MIGRATIONS):
        await migration.down(store)
        # Verify rollback

Deployment Coordination

Migration + Deploy Order

1. Deploy code that reads BOTH old and new format
2. Run migration (expand)
3. Deploy code that writes new format
4. Run migration (contract)
5. Deploy code that only reads new format

Feature Flags for Migration

class MigrationFeatureFlags:
    def __init__(self, flag_store):
        self.flags = flag_store

    def use_new_schema(self, tenant_id: str) -> bool:
        return self.flags.get(f"migration:{self.version}:enabled", tenant_id, default=False)

    async def enable_for_tenant(self, tenant_id: str):
        await self.flags.set(f"migration:{self.version}:enabled", tenant_id, True)

    async def enable_globally(self):
        await self.flags.set_global(f"migration:{self.version}:enabled", True)

Rollback Procedures

Automated Rollback

class AutoRollback:
    def __init__(self, migration_runner, health_checker):
        self.runner = migration_runner
        self.health = health_checker

    async def monitor_and_rollback(self, migration_version: str, 
                                   timeout: int = 300):
        start = time.time()
        while time.time() - start < timeout:
            health = await self.health.check_all()
            if not health.healthy:
                print(f"Health check failed, rolling back {migration_version}")
                await self.runner.rollback(migration_version)
                return
            await asyncio.sleep(10)

        print("Migration stabilized")

Related Notes

• Checkpointing — Checkpoint schema versioning
• Event Sourcing — Event schema evolution
• Session Management — Session schema migration
• Distributed State — Distributed migrations
• Deployment Architecture — Deploy coordination
• Reliability — Rollback procedures
• Observability — Migration monitoring
• Research Directions — Self-migrating systems

Sources: "Refactoring Databases" (Ambler, Sadalage), "Database Refactoring" patterns, "Zero-Downtime Migrations" (GitHub, Shopify, Stripe), Flyway, Liquibase, Django migrations, Alembic, temporal.io migrations, schema evolution best practices.