# 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
| Location | Purpose |
|---|---|
| **State schema_version** | Per-thread/aggregate version |
| **Event version** | Per-event schema version |
| **API version** | Client-facing contract |
| **Migration version** | Database/schema migration ID |

---

## 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
| Change | Compatible? |
|---|---|
| Add optional field | ✅ Yes |
| Add enum value | ✅ Yes |
| Add new event type | ✅ Yes |
| Add optional nested object | ✅ Yes |
| Increase array max length | ✅ Yes |
| Relax validation | ✅ Yes |
| Remove field | ❌ No |
| Change field type | ❌ No |
| Rename field | ❌ No |
| Make optional required | ❌ No |
| Change enum value | ❌ No |

---

## 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
```python
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
```python
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
```python
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
```python
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)
```python
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
```python
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
```python
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
```python
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
```python
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
```python
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
```python
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

- [[StateManagement/Checkpointing.md]] — Checkpoint schema versioning
- [[StateManagement/Event-Sourcing.md]] — Event schema evolution
- [[StateManagement/Session-Management.md]] — Session schema migration
- [[StateManagement/Distributed-State.md]] — Distributed migrations
- [[Production/Deployment-Architecture.md]] — Deploy coordination
- [[Production/Reliability.md]] — Rollback procedures
- [[Production/Observability.md]] — Migration monitoring
- [[Frontiers/Research-Directions.md]] — 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.*