diff --git a/src/langbot/pkg/persistence/alembic/versions/0009_add_workflow_tables.py b/src/langbot/pkg/persistence/alembic/versions/0009_add_workflow_tables.py new file mode 100644 index 000000000..f9fcb58fb --- /dev/null +++ b/src/langbot/pkg/persistence/alembic/versions/0009_add_workflow_tables.py @@ -0,0 +1,207 @@ +"""add workflow tables and bot binding fields + +Revision ID: 0009_add_workflow_tables +Revises: 0008_mcp_resource_prefs +Create Date: 2026-07-01 +""" + +from __future__ import annotations + +import sqlalchemy as sa +from alembic import op + +revision = '0009_add_workflow_tables' +down_revision = '0008_mcp_resource_prefs' +branch_labels = None +depends_on = None + + +def _table_exists(conn: sa.Connection, table_name: str) -> bool: + return table_name in sa.inspect(conn).get_table_names() + + +def _has_column(conn: sa.Connection, table_name: str, column_name: str) -> bool: + if not _table_exists(conn, table_name): + return False + return column_name in {column['name'] for column in sa.inspect(conn).get_columns(table_name)} + + +def _has_index_for_columns(conn: sa.Connection, table_name: str, columns: tuple[str, ...]) -> bool: + if not _table_exists(conn, table_name): + return False + for index in sa.inspect(conn).get_indexes(table_name): + if tuple(index.get('column_names') or ()) == columns: + return True + return False + + +def _ensure_index(conn: sa.Connection, table_name: str, index_name: str, columns: list[str]) -> None: + if _has_index_for_columns(conn, table_name, tuple(columns)): + return + op.create_index(index_name, table_name, columns) + + +def _create_workflow_tables(conn: sa.Connection) -> None: + if not _table_exists(conn, 'workflows'): + op.create_table( + 'workflows', + sa.Column('uuid', sa.String(255), primary_key=True), + sa.Column('name', sa.String(255), nullable=False), + sa.Column('description', sa.Text(), nullable=True), + sa.Column('emoji', sa.String(10), nullable=True), + sa.Column('version', sa.Integer(), nullable=False, server_default='1'), + sa.Column('is_enabled', sa.Boolean(), nullable=False, server_default=sa.true()), + sa.Column('definition', sa.JSON(), nullable=False, server_default=sa.text("'{}'")), + sa.Column('global_config', sa.JSON(), nullable=False, server_default=sa.text("'{}'")), + sa.Column( + 'extensions_preferences', + sa.JSON(), + nullable=False, + server_default=sa.text( + '\'{"enable_all_plugins": true, "enable_all_mcp_servers": true, "plugins": [], "mcp_servers": []}\'' + ), + ), + sa.Column('created_at', sa.DateTime(), nullable=False, server_default=sa.func.now()), + sa.Column('updated_at', sa.DateTime(), nullable=False, server_default=sa.func.now()), + ) + + if not _table_exists(conn, 'workflow_versions'): + op.create_table( + 'workflow_versions', + sa.Column('id', sa.Integer(), primary_key=True, autoincrement=True), + sa.Column('workflow_uuid', sa.String(255), nullable=False), + sa.Column('version', sa.Integer(), nullable=False), + sa.Column('definition', sa.JSON(), nullable=False), + sa.Column('global_config', sa.JSON(), nullable=False, server_default=sa.text("'{}'")), + sa.Column('created_at', sa.DateTime(), nullable=False, server_default=sa.func.now()), + sa.Column('created_by', sa.String(255), nullable=True), + sa.UniqueConstraint('workflow_uuid', 'version', name='uq_workflow_version'), + ) + + if not _table_exists(conn, 'workflow_triggers'): + op.create_table( + 'workflow_triggers', + sa.Column('uuid', sa.String(255), primary_key=True), + sa.Column('workflow_uuid', sa.String(255), nullable=False), + sa.Column('type', sa.String(50), nullable=False), + sa.Column('config', sa.JSON(), nullable=False, server_default=sa.text("'{}'")), + sa.Column('is_enabled', sa.Boolean(), nullable=False, server_default=sa.true()), + sa.Column('priority', sa.Integer(), nullable=False, server_default='0'), + sa.Column('created_at', sa.DateTime(), nullable=False, server_default=sa.func.now()), + sa.Column('updated_at', sa.DateTime(), nullable=False, server_default=sa.func.now()), + ) + + if not _table_exists(conn, 'workflow_executions'): + op.create_table( + 'workflow_executions', + sa.Column('uuid', sa.String(255), primary_key=True), + sa.Column('workflow_uuid', sa.String(255), nullable=False), + sa.Column('workflow_version', sa.Integer(), nullable=False), + sa.Column('status', sa.String(20), nullable=False), + sa.Column('trigger_type', sa.String(50), nullable=True), + sa.Column('trigger_data', sa.JSON(), nullable=True), + sa.Column('variables', sa.JSON(), nullable=True), + sa.Column('start_time', sa.DateTime(), nullable=True), + sa.Column('end_time', sa.DateTime(), nullable=True), + sa.Column('error', sa.Text(), nullable=True), + sa.Column('created_at', sa.DateTime(), nullable=False, server_default=sa.func.now()), + ) + + if not _table_exists(conn, 'workflow_node_executions'): + op.create_table( + 'workflow_node_executions', + sa.Column('id', sa.Integer(), primary_key=True, autoincrement=True), + sa.Column('execution_uuid', sa.String(255), nullable=False), + sa.Column('node_id', sa.String(100), nullable=False), + sa.Column('node_type', sa.String(50), nullable=False), + sa.Column('status', sa.String(20), nullable=False), + sa.Column('inputs', sa.JSON(), nullable=True), + sa.Column('outputs', sa.JSON(), nullable=True), + sa.Column('start_time', sa.DateTime(), nullable=True), + sa.Column('end_time', sa.DateTime(), nullable=True), + sa.Column('error', sa.Text(), nullable=True), + sa.Column('retry_count', sa.Integer(), nullable=False, server_default='0'), + ) + + if not _table_exists(conn, 'workflow_scheduled_jobs'): + op.create_table( + 'workflow_scheduled_jobs', + sa.Column('uuid', sa.String(255), primary_key=True), + sa.Column('trigger_uuid', sa.String(255), nullable=False), + sa.Column('cron_expression', sa.String(100), nullable=True), + sa.Column('next_run_time', sa.DateTime(), nullable=True), + sa.Column('last_run_time', sa.DateTime(), nullable=True), + sa.Column('is_enabled', sa.Boolean(), nullable=False, server_default=sa.true()), + ) + + _ensure_index(conn, 'workflow_versions', 'ix_workflow_versions_workflow_uuid', ['workflow_uuid']) + _ensure_index(conn, 'workflow_triggers', 'ix_workflow_triggers_workflow_uuid', ['workflow_uuid']) + _ensure_index(conn, 'workflow_executions', 'ix_workflow_executions_workflow_uuid', ['workflow_uuid']) + _ensure_index( + conn, + 'workflow_node_executions', + 'ix_workflow_node_executions_execution_uuid', + ['execution_uuid'], + ) + _ensure_index(conn, 'workflow_scheduled_jobs', 'ix_workflow_scheduled_jobs_trigger_uuid', ['trigger_uuid']) + + +def _add_bot_binding_fields(conn: sa.Connection) -> None: + if not _table_exists(conn, 'bots'): + return + + if not _has_column(conn, 'bots', 'binding_type'): + op.add_column( + 'bots', + sa.Column('binding_type', sa.String(32), nullable=False, server_default='pipeline'), + ) + + if not _has_column(conn, 'bots', 'binding_uuid'): + op.add_column('bots', sa.Column('binding_uuid', sa.String(64), nullable=True)) + + conn.execute( + sa.text(""" + UPDATE bots + SET binding_uuid = use_pipeline_uuid + WHERE use_pipeline_uuid IS NOT NULL + AND use_pipeline_uuid != '' + AND (binding_uuid IS NULL OR binding_uuid = '') + """) + ) + conn.execute( + sa.text(""" + UPDATE bots + SET binding_type = 'pipeline' + WHERE binding_uuid IS NOT NULL + AND binding_uuid != '' + AND (binding_type IS NULL OR binding_type = '') + """) + ) + + +def upgrade() -> None: + conn = op.get_bind() + _create_workflow_tables(conn) + _add_bot_binding_fields(conn) + + +def downgrade() -> None: + conn = op.get_bind() + + if _has_column(conn, 'bots', 'binding_uuid'): + with op.batch_alter_table('bots') as batch_op: + batch_op.drop_column('binding_uuid') + if _has_column(conn, 'bots', 'binding_type'): + with op.batch_alter_table('bots') as batch_op: + batch_op.drop_column('binding_type') + + for table_name in ( + 'workflow_scheduled_jobs', + 'workflow_node_executions', + 'workflow_executions', + 'workflow_triggers', + 'workflow_versions', + 'workflows', + ): + if _table_exists(conn, table_name): + op.drop_table(table_name) diff --git a/src/langbot/pkg/persistence/migrations/dbm026_workflow_tables.py b/src/langbot/pkg/persistence/migrations/dbm026_workflow_tables.py deleted file mode 100644 index 069d574f2..000000000 --- a/src/langbot/pkg/persistence/migrations/dbm026_workflow_tables.py +++ /dev/null @@ -1,158 +0,0 @@ -"""Add workflow tables and update bot binding fields""" - -import sqlalchemy -from .. import migration - - -@migration.migration_class(26) -class DBMigrateWorkflowTables(migration.DBMigration): - """Add workflow tables and update bot binding fields""" - - async def upgrade(self): - # Create workflows table - await self.ap.persistence_mgr.execute_async( - sqlalchemy.text(""" - CREATE TABLE IF NOT EXISTS workflows ( - uuid VARCHAR(255) PRIMARY KEY, - name VARCHAR(255) NOT NULL, - description TEXT, - emoji VARCHAR(10) DEFAULT '🔄', - version INTEGER NOT NULL DEFAULT 1, - is_enabled BOOLEAN NOT NULL DEFAULT 1, - definition JSON NOT NULL DEFAULT '{}', - global_config JSON NOT NULL DEFAULT '{}', - extensions_preferences JSON NOT NULL DEFAULT '{"enable_all_plugins": true, "enable_all_mcp_servers": true, "plugins": [], "mcp_servers": []}', - created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, - updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP - ) - """) - ) - - # Create workflow_versions table - await self.ap.persistence_mgr.execute_async( - sqlalchemy.text(""" - CREATE TABLE IF NOT EXISTS workflow_versions ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - workflow_uuid VARCHAR(255) NOT NULL, - version INTEGER NOT NULL, - definition JSON NOT NULL, - global_config JSON NOT NULL DEFAULT '{}', - created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, - created_by VARCHAR(255), - UNIQUE(workflow_uuid, version) - ) - """) - ) - - # Create workflow_triggers table - await self.ap.persistence_mgr.execute_async( - sqlalchemy.text(""" - CREATE TABLE IF NOT EXISTS workflow_triggers ( - uuid VARCHAR(255) PRIMARY KEY, - workflow_uuid VARCHAR(255) NOT NULL, - type VARCHAR(50) NOT NULL, - config JSON NOT NULL DEFAULT '{}', - is_enabled BOOLEAN NOT NULL DEFAULT 1, - priority INTEGER NOT NULL DEFAULT 0, - created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, - updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP - ) - """) - ) - - # Create workflow_executions table - await self.ap.persistence_mgr.execute_async( - sqlalchemy.text(""" - CREATE TABLE IF NOT EXISTS workflow_executions ( - uuid VARCHAR(255) PRIMARY KEY, - workflow_uuid VARCHAR(255) NOT NULL, - workflow_version INTEGER NOT NULL, - status VARCHAR(20) NOT NULL, - trigger_type VARCHAR(50), - trigger_data JSON, - variables JSON, - start_time TIMESTAMP, - end_time TIMESTAMP, - error TEXT, - created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP - ) - """) - ) - - # Create workflow_node_executions table - await self.ap.persistence_mgr.execute_async( - sqlalchemy.text(""" - CREATE TABLE IF NOT EXISTS workflow_node_executions ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - execution_uuid VARCHAR(255) NOT NULL, - node_id VARCHAR(100) NOT NULL, - node_type VARCHAR(50) NOT NULL, - status VARCHAR(20) NOT NULL, - inputs JSON, - outputs JSON, - start_time TIMESTAMP, - end_time TIMESTAMP, - error TEXT, - retry_count INTEGER NOT NULL DEFAULT 0 - ) - """) - ) - - # Create workflow_scheduled_jobs table - await self.ap.persistence_mgr.execute_async( - sqlalchemy.text(""" - CREATE TABLE IF NOT EXISTS workflow_scheduled_jobs ( - uuid VARCHAR(255) PRIMARY KEY, - trigger_uuid VARCHAR(255) NOT NULL, - cron_expression VARCHAR(100), - next_run_time TIMESTAMP, - last_run_time TIMESTAMP, - is_enabled BOOLEAN NOT NULL DEFAULT 1 - ) - """) - ) - - # Create indexes - await self.ap.persistence_mgr.execute_async( - sqlalchemy.text('CREATE INDEX IF NOT EXISTS idx_workflow_versions_uuid ON workflow_versions(workflow_uuid)') - ) - await self.ap.persistence_mgr.execute_async( - sqlalchemy.text('CREATE INDEX IF NOT EXISTS idx_workflow_triggers_uuid ON workflow_triggers(workflow_uuid)') - ) - await self.ap.persistence_mgr.execute_async( - sqlalchemy.text( - 'CREATE INDEX IF NOT EXISTS idx_workflow_executions_uuid ON workflow_executions(workflow_uuid)' - ) - ) - await self.ap.persistence_mgr.execute_async( - sqlalchemy.text( - 'CREATE INDEX IF NOT EXISTS idx_workflow_node_executions_uuid ON workflow_node_executions(execution_uuid)' - ) - ) - await self.ap.persistence_mgr.execute_async( - sqlalchemy.text( - 'CREATE INDEX IF NOT EXISTS idx_workflow_scheduled_jobs_trigger ON workflow_scheduled_jobs(trigger_uuid)' - ) - ) - - # Update bots table: add binding_type column (default to 'pipeline' for backward compatibility) - # Check if column exists first (SQLite doesn't support IF NOT EXISTS for columns) - try: - await self.ap.persistence_mgr.execute_async(sqlalchemy.text('SELECT binding_type FROM bots LIMIT 1')) - except Exception: - # Column doesn't exist, add it - await self.ap.persistence_mgr.execute_async( - sqlalchemy.text("ALTER TABLE bots ADD COLUMN binding_type VARCHAR(20) NOT NULL DEFAULT 'pipeline'") - ) - - async def downgrade(self): - # Drop tables in reverse order - await self.ap.persistence_mgr.execute_async(sqlalchemy.text('DROP TABLE IF EXISTS workflow_scheduled_jobs')) - await self.ap.persistence_mgr.execute_async(sqlalchemy.text('DROP TABLE IF EXISTS workflow_node_executions')) - await self.ap.persistence_mgr.execute_async(sqlalchemy.text('DROP TABLE IF EXISTS workflow_executions')) - await self.ap.persistence_mgr.execute_async(sqlalchemy.text('DROP TABLE IF EXISTS workflow_triggers')) - await self.ap.persistence_mgr.execute_async(sqlalchemy.text('DROP TABLE IF EXISTS workflow_versions')) - await self.ap.persistence_mgr.execute_async(sqlalchemy.text('DROP TABLE IF EXISTS workflows')) - - # Remove binding_type column from bots (SQLite doesn't support DROP COLUMN directly) - # This would need a table recreation in SQLite, so we'll skip it in downgrade diff --git a/src/langbot/pkg/persistence/migrations/dbm027_bot_binding_fields.py b/src/langbot/pkg/persistence/migrations/dbm027_bot_binding_fields.py deleted file mode 100644 index 5de8aae39..000000000 --- a/src/langbot/pkg/persistence/migrations/dbm027_bot_binding_fields.py +++ /dev/null @@ -1,49 +0,0 @@ -"""Add binding_uuid field to bots table and migrate data""" - -import sqlalchemy -from .. import migration - - -@migration.migration_class(27) -class DBMigrateBotBindingFields(migration.DBMigration): - """Add binding_uuid field to bots table and migrate existing data""" - - async def upgrade(self): - # Add binding_uuid column to bots table - # Check if column exists first (SQLite doesn't support IF NOT EXISTS for columns) - try: - await self.ap.persistence_mgr.execute_async(sqlalchemy.text('SELECT binding_uuid FROM bots LIMIT 1')) - except Exception: - # Column doesn't exist, add it - await self.ap.persistence_mgr.execute_async( - sqlalchemy.text('ALTER TABLE bots ADD COLUMN binding_uuid VARCHAR(64)') - ) - - # Migrate existing data: copy use_pipeline_uuid to binding_uuid for records - # that have a pipeline bound and binding_uuid is not set yet - await self.ap.persistence_mgr.execute_async( - sqlalchemy.text(""" - UPDATE bots - SET binding_uuid = use_pipeline_uuid - WHERE use_pipeline_uuid IS NOT NULL - AND use_pipeline_uuid != '' - AND (binding_uuid IS NULL OR binding_uuid = '') - """) - ) - - # Ensure binding_type is 'pipeline' for records that were migrated - await self.ap.persistence_mgr.execute_async( - sqlalchemy.text(""" - UPDATE bots - SET binding_type = 'pipeline' - WHERE binding_uuid IS NOT NULL - AND binding_uuid != '' - AND (binding_type IS NULL OR binding_type = '') - """) - ) - - async def downgrade(self): - # SQLite doesn't support DROP COLUMN directly - # This would need a table recreation in SQLite, so we'll skip it in downgrade - # The column will remain but won't be used - pass