Merge master into dev/4.11.x and preserve plugin runner architecture

Reconcile migration branches without rewriting published revisions; retain additive Codex, monitoring, provider and platform fixes. Keep dynamic runner schemas and Host ownership, restore compatibility regressions, and preserve safe model-test error handling.

src/langbot/pkg/persistence/alembic/versions/0022_codex_credentials.py

@@ -0,0 +1,48 @@
+"""Add isolated server-only Codex credentials and tenant RLS.
+
+Revision ID: 0022_codex_credentials
+Revises: 0021_merge_reasoning_config
+"""
+
+from alembic import op
+import sqlalchemy as sa
+
+revision = '0022_codex_credentials'
+down_revision = '0021_merge_reasoning_config'
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    conn = op.get_bind()
+    # Fresh startup creates ORM metadata before running Alembic.
+    if 'codex_credentials' not in sa.inspect(conn).get_table_names():
+        op.create_table(
+            'codex_credentials',
+            sa.Column('provider_uuid', sa.String(255), primary_key=True),
+            sa.Column('workspace_uuid', sa.String(36), nullable=False),
+            sa.Column('payload', sa.JSON(), nullable=False),
+            sa.Column('version', sa.Integer(), nullable=False),
+            sa.Column('lease_owner', sa.String(64), nullable=True),
+            sa.Column('lease_until', sa.Float(), nullable=False),
+            sa.ForeignKeyConstraint(
+                ['workspace_uuid', 'provider_uuid'],
+                ['model_providers.workspace_uuid', 'model_providers.uuid'],
+                name='fk_codex_credentials_workspace_provider',
+                ondelete='CASCADE',
+            ),
+        )
+        op.create_index('ix_codex_credentials_workspace', 'codex_credentials', ['workspace_uuid'])
+    if conn.dialect.name == 'postgresql':
+        op.execute('ALTER TABLE codex_credentials ENABLE ROW LEVEL SECURITY')
+        op.execute('ALTER TABLE codex_credentials FORCE ROW LEVEL SECURITY')
+        op.execute('DROP POLICY IF EXISTS langbot_workspace_isolation ON codex_credentials')
+        expression = "workspace_uuid::text = NULLIF(current_setting('langbot.workspace_uuid', true), '')"
+        op.execute(
+            f'CREATE POLICY langbot_workspace_isolation ON codex_credentials '
+            f'FOR ALL USING ({expression}) WITH CHECK ({expression})'
+        )
+
+
+def downgrade() -> None:
+    op.drop_table('codex_credentials')

src/langbot/pkg/persistence/alembic/versions/0023_bot_scoped_sessions.py

@@ -0,0 +1,104 @@
+"""Scope monitoring sessions by bot without changing runtime session IDs.
+
+Revision ID: 0023_bot_scoped_sessions
+Revises: 0022_codex_credentials
+"""
+
+from alembic import op
+import sqlalchemy as sa
+from sqlalchemy.dialects import postgresql, sqlite
+
+revision = '0023_bot_scoped_sessions'
+down_revision = '0022_codex_credentials'
+branch_labels = None
+depends_on = None
+
+_TABLE = 'monitoring_sessions'
+_KEY = ['workspace_uuid', 'bot_id', 'session_id']
+
+
+def upgrade() -> None:
+    conn = op.get_bind()
+    inspector = sa.inspect(conn)
+    if _TABLE not in inspector.get_table_names():
+        return
+    pk = inspector.get_pk_constraint(_TABLE)
+    if pk['constrained_columns'] == _KEY:
+        return
+    # PostgreSQL alters in place, retaining indexes, grants, policies and RLS.
+    # SQLite batch reflection retains all existing indexes and foreign keys.
+    with op.batch_alter_table(_TABLE, naming_convention={'pk': 'pk_%(table_name)s'}) as batch:
+        batch.drop_constraint(pk['name'] or f'pk_{_TABLE}', type_='primary')
+        batch.create_primary_key(f'pk_{_TABLE}', _KEY)
+
+    metadata = sa.MetaData()
+    sessions = sa.Table(_TABLE, metadata, autoload_with=conn)
+    messages = sa.Table('monitoring_messages', metadata, autoload_with=conn)
+    m = messages.c
+    collisions = (
+        sa.select(m.workspace_uuid, m.session_id)
+        .group_by(m.workspace_uuid, m.session_id)
+        .having(sa.func.count(sa.distinct(m.bot_id)) > 1)
+        .subquery()
+    )
+    partition = [m.workspace_uuid, m.bot_id, m.session_id]
+    # Repair only demonstrable collisions. Retention may have removed earlier
+    # evidence; these summaries describe surviving messages, never invented text.
+    ranked = (
+        sa.select(
+            *[m[name] for name in _KEY],
+            m.bot_name,
+            m.pipeline_id,
+            m.pipeline_name,
+            m.platform,
+            m.user_id,
+            m.user_name,
+            sa.func.sum(sa.case((sa.or_(m.role == 'user', m.role.is_(None)), 1), else_=0))
+            .over(partition_by=partition)
+            .label('message_count'),
+            sa.func.min(m.timestamp).over(partition_by=partition).label('start_time'),
+            sa.func.max(m.timestamp).over(partition_by=partition).label('last_activity'),
+            sa.func.row_number().over(partition_by=partition, order_by=[m.timestamp.desc(), m.id.desc()]).label('rank'),
+        )
+        .join(
+            collisions,
+            sa.and_(m.workspace_uuid == collisions.c.workspace_uuid, m.session_id == collisions.c.session_id),
+        )
+        .subquery()
+    )
+    columns = _KEY + [
+        'bot_name',
+        'pipeline_id',
+        'pipeline_name',
+        'platform',
+        'user_id',
+        'user_name',
+        'message_count',
+        'start_time',
+        'last_activity',
+        'is_active',
+    ]
+    select = sa.select(*[ranked.c[name] for name in columns[:-1]], sa.literal(True)).where(ranked.c.rank == 1)
+    insert = postgresql.insert if conn.dialect.name == 'postgresql' else sqlite.insert
+    statement = insert(sessions).from_select(columns, select)
+    conn.execute(
+        statement.on_conflict_do_update(
+            index_elements=_KEY,
+            set_={name: statement.excluded[name] for name in columns if name not in _KEY and name != 'is_active'},
+        )
+    )
+
+
+def downgrade() -> None:
+    conn = op.get_bind()
+    if _TABLE not in sa.inspect(conn).get_table_names():
+        return
+    collisions = conn.execute(
+        sa.text('SELECT 1 FROM monitoring_sessions GROUP BY workspace_uuid, session_id HAVING COUNT(*) > 1 LIMIT 1')
+    ).first()
+    if collisions:
+        raise RuntimeError('Cannot downgrade bot-scoped sessions without losing colliding bot records')
+    pk = sa.inspect(conn).get_pk_constraint(_TABLE)
+    with op.batch_alter_table(_TABLE, naming_convention={'pk': 'pk_%(table_name)s'}) as batch:
+        batch.drop_constraint(pk['name'] or f'pk_{_TABLE}', type_='primary')
+        batch.create_primary_key(f'pk_{_TABLE}', ['workspace_uuid', 'session_id'])

src/langbot/pkg/persistence/alembic/versions/0024_passkey_credentials.py

@@ -0,0 +1,54 @@
+"""add passkey credentials table
+
+Revision ID: 0024_passkey_credentials
+Revises: 0023_bot_scoped_sessions
+Create Date: 2026-09-12
+"""
+
+from __future__ import annotations
+
+import sqlalchemy as sa
+from alembic import op
+
+revision = '0024_passkey_credentials'
+down_revision = '0023_bot_scoped_sessions'
+branch_labels = None
+depends_on = None
+
+_TABLE_NAME = 'passkey_credentials'
+
+
+def upgrade() -> None:
+    conn = op.get_bind()
+    existing_tables = set(sa.inspect(conn).get_table_names())
+    if _TABLE_NAME not in existing_tables:
+        op.create_table(
+            _TABLE_NAME,
+            sa.Column('id', sa.Integer(), primary_key=True, autoincrement=True),
+            sa.Column('uuid', sa.String(36), nullable=False),
+            sa.Column(
+                'account_uuid',
+                sa.String(36),
+                sa.ForeignKey('users.uuid', ondelete='CASCADE'),
+                nullable=False,
+            ),
+            sa.Column('name', sa.String(255), nullable=False),
+            sa.Column('credential_id', sa.String(255), nullable=False),
+            sa.Column('public_key', sa.Text(), nullable=False),
+            sa.Column('sign_count', sa.Integer(), nullable=False, server_default='0'),
+            sa.Column('aaguid', sa.String(64), nullable=True),
+            sa.Column('transports', sa.String(255), nullable=True),
+            sa.Column('backed_up', sa.Boolean(), nullable=False, server_default='0'),
+            sa.Column('created_at', sa.DateTime(), nullable=False, server_default=sa.func.now()),
+            sa.Column('last_used_at', sa.DateTime(), nullable=True),
+        )
+        op.create_index('uq_passkey_credentials_uuid', _TABLE_NAME, ['uuid'], unique=True)
+        op.create_index('uq_passkey_credentials_cred_id', _TABLE_NAME, ['credential_id'], unique=True)
+        op.create_index('ix_passkey_credentials_account', _TABLE_NAME, ['account_uuid'], unique=False)
+
+
+def downgrade() -> None:
+    op.drop_index('ix_passkey_credentials_account', table_name=_TABLE_NAME)
+    op.drop_index('uq_passkey_credentials_cred_id', table_name=_TABLE_NAME)
+    op.drop_index('uq_passkey_credentials_uuid', table_name=_TABLE_NAME)
+    op.drop_table(_TABLE_NAME)

src/langbot/pkg/persistence/alembic/versions/0026_merge_master_beta.py

@@ -0,0 +1,22 @@
+"""Join the published master credentials and 4.11 bot processor branches.
+
+Revision ID: 0026_merge_master_beta
+Revises: 0024_passkey_credentials, 0025_bot_plugin_processors
+
+Both parent chains are already published. Keep their identities and schema/data
+operations intact; Alembic applies the missing branch before this no-op merge.
+"""
+
+revision = '0026_merge_master_beta'
+down_revision = ('0024_passkey_credentials', '0025_bot_plugin_processors')
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    pass
+
+
+def downgrade() -> None:
+    # Only unmerge the revision graph; neither branch's schema/data is removed.
+    pass

src/langbot/pkg/persistence/mgr.py

@@ -64,6 +64,8 @@ _ALEMBIC_TENANT_TABLES = {
     'binary_storages',
     'mcp_servers',
     'model_providers',
+    'codex_credentials',
+    'passkey_credentials',
     'llm_models',
     'embedding_models',
     'rerank_models',

src/langbot/pkg/persistence/tenant_uow.py

@@ -52,6 +52,7 @@ TENANT_TABLE_COLUMNS: dict[str, str] = {
     'binary_storages': 'workspace_uuid',
     'mcp_servers': 'workspace_uuid',
     'model_providers': 'workspace_uuid',
+    'codex_credentials': 'workspace_uuid',
     'llm_models': 'workspace_uuid',
     'embedding_models': 'workspace_uuid',
     'rerank_models': 'workspace_uuid',
@@ -207,6 +208,8 @@ _SYNC_PROXY_CAPABILITY: contextvars.ContextVar[_ScopedSessionGuardState | None]
 _ALLOWED_SCOPED_BUILTIN_FUNCTION_TYPES = {
     'coalesce': sqlalchemy.sql.functions.coalesce,
     'count': sqlalchemy.sql.functions.count,
+    'min': sqlalchemy.sql.functions.min,
+    'max': sqlalchemy.sql.functions.max,
     'now': sqlalchemy.sql.functions.now,
     'sum': sqlalchemy.sql.functions.sum,
 }
@@ -853,7 +856,30 @@ class TenantScopedAsyncSession(sqlalchemy_asyncio.AsyncSession):
         self._require_owner_task()
         self._enter_internal_access()
         try:
-            await transaction.commit()
+            # Retain the actual connection before COMMIT: after a failed SQLite
+            # COMMIT the logical transaction is inactive, but the DBAPI writer
+            # can still hold PENDING/RESERVED locks. Session.close()/rollback()
+            # alone can then return that poisoned connection to the pool.
+            connection = await super().connection()
+            try:
+                await transaction.commit()
+            except BaseException as exc:
+                cleanup = asyncio.create_task(connection.invalidate())
+                # Invalidation does not access the task-owned Session. Shield
+                # physical cleanup, including against repeated cancellation,
+                # before the owner closes the Session and releases its scope.
+                while not cleanup.done():
+                    try:
+                        await asyncio.shield(cleanup)
+                    except asyncio.CancelledError:
+                        continue
+                    except BaseException:
+                        break
+                try:
+                    cleanup.result()
+                except BaseException as cleanup_error:
+                    exc.add_note(f'Failed to invalidate transaction connection: {cleanup_error!r}')
+                raise
         finally:
             self._exit_internal_access()
 
@@ -1370,6 +1396,7 @@ class TenantUnitOfWork:
             state.mark_rollback_only(exc_value)
         rollback_only = state.rollback_only
         committed = False
+        transaction_error: BaseException | None = None
         try:
             if exc_type is None and not rollback_only:
                 await typing.cast(TenantScopedAsyncSession, session)._commit_owned_transaction(
@@ -1382,6 +1409,9 @@ class TenantUnitOfWork:
                     _UOW_SESSION_CONTROL_CAPABILITY,
                     transaction,
                 )
+        except BaseException as exc:
+            transaction_error = exc
+            raise
         finally:
             try:
                 if self._active_transaction is not None and self._context_token is not None:
@@ -1389,6 +1419,10 @@ class TenantUnitOfWork:
                 await typing.cast(TenantScopedAsyncSession, session)._close_owned_session(
                     _UOW_SESSION_CONTROL_CAPABILITY
                 )
+            except BaseException as cleanup_error:
+                if transaction_error is None:
+                    raise
+                transaction_error.add_note(f'Failed to close transaction Session: {cleanup_error!r}')
             finally:
                 if self._database_operation_token is not None:
                     _DATABASE_OPERATION_TRANSACTION.reset(self._database_operation_token)