"""Phase 5 migration for MAF external-write target leases. Creates the additive ``maf_roundtable_action_leases`` table used to serialize approved writes to the same server-derived external business target. Existing action rows are intentionally not backfilled: old rows have no trusted target key and retain the legacy action-level at-most-once behavior. New writes receive an opaque target key from ``ActionCoordinator``. Run this before deploying code that enables MAF external writes: PYTHONPATH=. uv run python scripts/migrate_maf_roundtable_phase5.py --dry-run PYTHONPATH=. uv run python scripts/migrate_maf_roundtable_phase5.py """ from __future__ import annotations import argparse import asyncio import logging import sys from typing import Any logging.basicConfig(level=logging.INFO, format="%(message)s") logger = logging.getLogger("migrate_maf_roundtable_phase5") TABLE = "maf_roundtable_action_leases" ACTION_INDEX = "ix_maf_action_leases_action_id" EXPIRY_INDEX = "ix_maf_action_leases_lease_expires_at" def _dialect_name(conn_or_dialect: Any) -> str: return str(getattr(getattr(conn_or_dialect, "dialect", conn_or_dialect), "name", "sqlite")).lower() def _timestamp_sql(conn_or_dialect: Any) -> str: """Return the timestamp type accepted by each supported database.""" dialect = _dialect_name(conn_or_dialect) if dialect == "postgresql": return "TIMESTAMP WITH TIME ZONE" if dialect in {"mysql", "mariadb"}: return "DATETIME(6)" return "DATETIME" def _create_lease_table_sql(conn_or_dialect: Any) -> str: timestamp_type = _timestamp_sql(conn_or_dialect) return f""" CREATE TABLE {TABLE} ( target_key VARCHAR(255) NOT NULL PRIMARY KEY, action_id VARCHAR(64) NOT NULL, lease_expires_at {timestamp_type} NOT NULL, created_at {timestamp_type} NOT NULL, updated_at {timestamp_type} NOT NULL, CONSTRAINT uq_maf_action_leases_action UNIQUE (action_id), FOREIGN KEY (action_id) REFERENCES maf_roundtable_actions (id) ON DELETE CASCADE ) """ def validate_lease_table_schema(inspector: Any) -> list[str]: """Return every lease-table contract violation found by SQLAlchemy.""" if TABLE not in inspector.get_table_names(): return [f"missing table: {TABLE}"] required_columns = ("target_key", "action_id", "lease_expires_at", "created_at", "updated_at") columns = {column["name"]: column for column in inspector.get_columns(TABLE)} issues: list[str] = [] for name in required_columns: column = columns.get(name) if column is None: issues.append(f"missing column: {name}") elif column.get("nullable") is not False: issues.append(f"nullable column: {name}") primary_key = inspector.get_pk_constraint(TABLE).get("constrained_columns") or [] if list(primary_key) != ["target_key"]: issues.append(f"primary key must be (target_key), found {list(primary_key)}") foreign_key_ok = False for foreign_key in inspector.get_foreign_keys(TABLE): options = foreign_key.get("options") or {} if ( list(foreign_key.get("constrained_columns") or []) == ["action_id"] and foreign_key.get("referred_table") == "maf_roundtable_actions" and list(foreign_key.get("referred_columns") or []) == ["id"] and str(options.get("ondelete") or "").upper() == "CASCADE" ): foreign_key_ok = True break if not foreign_key_ok: issues.append( "missing foreign key: action_id -> maf_roundtable_actions.id ON DELETE CASCADE" ) unique_action_id = any( list(constraint.get("column_names") or []) == ["action_id"] for constraint in inspector.get_unique_constraints(TABLE) ) # Some dialects expose a named UNIQUE constraint only as a unique index. unique_action_id = unique_action_id or any( bool(index.get("unique")) and list(index.get("column_names") or []) == ["action_id"] for index in inspector.get_indexes(TABLE) ) if not unique_action_id: issues.append("missing unique constraint: action_id") indexes = {index.get("name"): list(index.get("column_names") or []) for index in inspector.get_indexes(TABLE)} for name, column in ((ACTION_INDEX, "action_id"), (EXPIRY_INDEX, "lease_expires_at")): if name not in indexes: issues.append(f"missing index: {name}") elif indexes[name] != [column]: issues.append(f"invalid index: {name} must cover ({column})") return issues def _table_has_rows(conn: Any, text: Any) -> bool: return conn.execute(text(f"SELECT 1 FROM {TABLE} LIMIT 1")).first() is not None def _create_indexes(execute: Any) -> None: execute(f"CREATE INDEX {ACTION_INDEX} ON {TABLE} (action_id)") execute(f"CREATE INDEX {EXPIRY_INDEX} ON {TABLE} (lease_expires_at)") async def _migrate(dry_run: bool) -> int: from deerflow.config import get_app_config from deerflow.persistence.engine import close_engine, get_session_factory, init_engine_from_config await init_engine_from_config(get_app_config().database) sf = get_session_factory() if sf is None: logger.error("Database backend is 'memory' — nothing to migrate.") return 0 engine = sf.kw["bind"] async with engine.begin() as conn: await conn.run_sync(_migrate_sync, dry_run=dry_run) await close_engine() return 0 def _migrate_sync(conn: Any, *, dry_run: bool) -> None: from sqlalchemy import inspect, text inspector = inspect(conn) def execute(sql: str) -> None: if dry_run: logger.info("[dry-run] would execute: %s", " ".join(sql.split())) return conn.execute(text(sql)) if TABLE not in inspector.get_table_names(): logger.info("Creating %s", TABLE) execute(_create_lease_table_sql(conn)) _create_indexes(execute) if dry_run: return inspector = inspect(conn) else: logger.info("Table %s already exists", TABLE) issues = validate_lease_table_schema(inspector) constraint_issues = [ issue for issue in issues if not issue.startswith("missing index:") ] if constraint_issues: has_rows = _table_has_rows(conn, text) detail = "; ".join(constraint_issues) if has_rows: raise RuntimeError( f"{TABLE} has data but its lease safety schema is incomplete: {detail}. " "Back up and manually rebuild the table after operational review." ) logger.warning( "%s is empty but incomplete; rebuilding it with the current full schema: %s", TABLE, detail, ) execute(f"DROP TABLE {TABLE}") execute(_create_lease_table_sql(conn)) _create_indexes(execute) if dry_run: return inspector = inspect(conn) else: for index_name, column in ( (ACTION_INDEX, "action_id"), (EXPIRY_INDEX, "lease_expires_at"), ): if f"missing index: {index_name}" in issues: execute(f"CREATE INDEX {index_name} ON {TABLE} ({column})") if dry_run: return inspector = inspect(conn) final_issues = validate_lease_table_schema(inspector) if final_issues: raise RuntimeError( f"{TABLE} migration did not produce the required lease schema: " + "; ".join(final_issues) ) def main() -> int: parser = argparse.ArgumentParser(description="Create MAF external-write target lease table") parser.add_argument("--dry-run", action="store_true", help="Preview only; do not write") return asyncio.run(_migrate(parser.parse_args().dry_run)) if __name__ == "__main__": sys.exit(main())