214 lines
7.9 KiB
Python
214 lines
7.9 KiB
Python
"""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())
|