diff --git a/backend/open_webui/migrations/versions/f8c0e5b134cd_add_tool_history_and_function_history.py b/backend/open_webui/migrations/versions/f8c0e5b134cd_add_tool_history_and_function_history.py new file mode 100644 index 0000000000..d703417a56 --- /dev/null +++ b/backend/open_webui/migrations/versions/f8c0e5b134cd_add_tool_history_and_function_history.py @@ -0,0 +1,64 @@ +"""Add Tool and Function history without executing or rewriting plugin source.""" + +import json +import uuid + +import sqlalchemy as sa +from alembic import op + +revision = 'f8c0e5b134cd' +down_revision = 'e7b9d4a023bc' +branch_labels = None +depends_on = None + + +def upgrade(): + connection = op.get_bind() + for kind in ('tool', 'function'): + op.add_column(kind, sa.Column('version_id', sa.Text(), nullable=True)) + history = op.create_table( + f'{kind}_history', + sa.Column('id', sa.Text(), primary_key=True), + sa.Column(f'{kind}_id', sa.Text(), nullable=False), + sa.Column('parent_id', sa.Text(), nullable=True), + sa.Column('snapshot', sa.JSON(), nullable=False), + sa.Column('user_id', sa.Text(), nullable=False), + sa.Column('commit_message', sa.Text(), nullable=True), + sa.Column('created_at', sa.BigInteger(), nullable=False), + ) + op.create_index(f'ix_{kind}_history_{kind}_id', f'{kind}_history', [f'{kind}_id']) + table = sa.Table(kind, sa.MetaData(), autoload_with=connection) + for row in connection.execute(sa.select(table)).mappings(): + meta = row['meta'] or {} + meta = json.loads(meta) if isinstance(meta, str) else dict(meta) + for key in ( + ('manifest', 'has_user_valves', 'toggle') if kind == 'function' else ('manifest', 'has_user_valves') + ): + meta.pop(key, None) + meta.setdefault('description', None) + if not meta.get('i18n'): + meta.pop('i18n', None) + snapshot = {'name': row['name'], 'content': row['content'] or '', 'meta': meta} + version_id = str(uuid.uuid4()) + connection.execute( + history.insert().values( + **{ + 'id': version_id, + f'{kind}_id': row['id'], + 'parent_id': None, + 'snapshot': snapshot, + 'user_id': row['user_id'] or '', + 'commit_message': None, + 'created_at': row['updated_at'] or row['created_at'] or 0, + } + ) + ) + connection.execute(table.update().where(table.c.id == row['id']).values(version_id=version_id)) + + +def downgrade(): + for kind in ('function', 'tool'): + op.drop_index(f'ix_{kind}_history_{kind}_id', table_name=f'{kind}_history') + op.drop_table(f'{kind}_history') + with op.batch_alter_table(kind) as batch: + batch.drop_column('version_id') diff --git a/backend/open_webui/models/function_history.py b/backend/open_webui/models/function_history.py new file mode 100644 index 0000000000..47dd401aea --- /dev/null +++ b/backend/open_webui/models/function_history.py @@ -0,0 +1,162 @@ +"""Immutable snapshots of function configuration; the live row remains Production.""" + +import difflib +import time +import uuid +from copy import deepcopy + +from fastapi import HTTPException +from open_webui.internal.db import Base, get_async_db_context +from pydantic import BaseModel, ConfigDict +from sqlalchemy import JSON, BigInteger, Column, Text, select, update + + +class FunctionHistory(Base): + __tablename__ = 'function_history' + id = Column(Text, primary_key=True) + function_id = Column(Text, nullable=False, index=True) + parent_id = Column(Text, nullable=True) + snapshot = Column(JSON, nullable=False) + user_id = Column(Text, nullable=False) + commit_message = Column(Text, nullable=True) + created_at = Column(BigInteger, nullable=False) + + +class FunctionHistoryResponse(BaseModel): + model_config = ConfigDict(from_attributes=True) + id: str + function_id: str + parent_id: str | None = None + user_id: str + commit_message: str | None = None + created_at: int + user: dict | None = None + + +class FunctionHistoryModel(FunctionHistoryResponse): + snapshot: dict + + +class FunctionHistoryTable: + async def delete_history_entry(self, function_id, history_id, db=None): + from open_webui.models.functions import Function + + async with get_async_db_context(db) as session: + try: + # Serialize with production switches on both SQLite and PostgreSQL. + await session.execute( + update(Function).where(Function.id == function_id).values(version_id=Function.version_id) + ) + model = await session.get(Function, function_id, populate_existing=True) + if not model: + return False + if model.version_id == history_id: + raise HTTPException(400, 'Cannot delete the current version') + entry = ( + await session.execute(select(FunctionHistory).filter_by(id=history_id, function_id=function_id)) + ).scalar_one_or_none() + if not entry: + return False + await session.execute( + update(FunctionHistory) + .where(FunctionHistory.function_id == function_id, FunctionHistory.parent_id == history_id) + .values(parent_id=entry.parent_id) + ) + await session.delete(entry) + await session.commit() + return True + except Exception: + await session.rollback() + raise + + def new_entry(self, function_id, snapshot, user_id, parent_id=None, commit_message=None): + return FunctionHistory( + id=str(uuid.uuid4()), + function_id=function_id, + snapshot=snapshot, + user_id=user_id, + parent_id=parent_id, + commit_message=commit_message, + created_at=int(time.time()), + ) + + async def get_history_by_id(self, function_id, history_id, db=None): + from open_webui.models.users import User + + async with get_async_db_context(db) as session: + entry = ( + await session.execute(select(FunctionHistory).filter_by(function_id=function_id, id=history_id)) + ).scalar_one_or_none() + if not entry: + return None + result = FunctionHistoryModel.model_validate(entry) + author = (await session.execute(select(User.name).where(User.id == entry.user_id))).scalar_one_or_none() + result.user = {'name': author} if author else None + return result + + async def get_history_by_function_id(self, function_id, page=1, db=None): + from open_webui.models.users import User + + async with get_async_db_context(db) as session: + columns = [getattr(FunctionHistory, key) for key in FunctionHistoryResponse.model_fields if key != 'user'] + rows = ( + ( + await session.execute( + select(*columns, User.name.label('author_name')) + .outerjoin(User, User.id == FunctionHistory.user_id) + .where(FunctionHistory.function_id == function_id) + .order_by(FunctionHistory.created_at.desc(), FunctionHistory.id.desc()) + .offset((max(1, page) - 1) * 20) + .limit(20) + ) + ) + .mappings() + .all() + ) + return [ + FunctionHistoryResponse( + **{key: value for key, value in row.items() if key != 'author_name'}, + user={'name': row['author_name']} if row['author_name'] else None, + ) + for row in rows + ] + + +FunctionHistories = FunctionHistoryTable() + + +def function_snapshot(resource): + data = ( + resource if isinstance(resource, dict) else {key: getattr(resource, key) for key in ('name', 'content', 'meta')} + ) + meta = data.get('meta') or {} + meta = meta.model_dump() if isinstance(meta, BaseModel) else deepcopy(meta) + meta.setdefault('description', None) + if not meta.get('i18n'): + meta.pop('i18n', None) + for key in ('manifest', 'has_user_valves', 'toggle'): + meta.pop(key, None) + return {'name': data.get('name'), 'content': data.get('content') or '', 'meta': meta} + + +def function_diff(before, after): + left, right = before.snapshot, after.snapshot + metadata = { + key: {'before': left.get(key), 'after': right.get(key)} + for key in ('name', 'meta') + if left.get(key) != right.get(key) + } + old, new = left.get('content') or '', right.get('content') or '' + # splitlines handles a missing final newline and CRLF without breaking the renderer. + patch = '\n'.join( + difflib.unified_diff( + old.splitlines(), new.splitlines(), fromfile='selected.py', tofile='production.py', lineterm='' + ) + ) + return { + 'from_id': before.id, + 'to_id': after.id, + 'metadata': metadata, + 'content_diff': patch, + 'line_endings_only': old != new and not patch, + } diff --git a/backend/open_webui/models/functions.py b/backend/open_webui/models/functions.py index 573880d521..a06f2bac17 100644 --- a/backend/open_webui/models/functions.py +++ b/backend/open_webui/models/functions.py @@ -6,9 +6,11 @@ import logging import time # local imports +from fastapi import HTTPException from open_webui.internal.db import Base, JSONField, get_async_db_context +from open_webui.models.function_history import FunctionHistories, FunctionHistory, function_snapshot from open_webui.models.users import User, UserResponse, Users, UserSettings -from open_webui.utils.valves import decrypt_valves, encrypt_valves +from open_webui.utils.valves import decrypt_valves, encrypt_valves, validate_valves from pydantic import BaseModel, ConfigDict from sqlalchemy import BigInteger, Boolean, Column, Index, String, Text, delete, select, update from sqlalchemy.ext.asyncio import AsyncSession @@ -20,6 +22,7 @@ class Function(Base): # database table mapping __tablename__ = 'function' id = Column(String, primary_key=True, unique=True) + version_id = Column(Text, nullable=True) user_id = Column(String, index=True) # creator user id name = Column(Text, nullable=False) # function identifier type = Column(Text, nullable=False) # function type (pipe, filter, etc.) @@ -41,6 +44,7 @@ class FunctionMeta(BaseModel): class FunctionModel(BaseModel): + version_id: str | None = None id: str user_id: str | None = None # may be null for legacy/malformed records name: str @@ -57,6 +61,7 @@ class FunctionModel(BaseModel): # --- form / schema definitions --- class FunctionWithValvesModel(BaseModel): + version_id: str | None = None id: str user_id: str | None = None # may be null for legacy/malformed records name: str @@ -78,6 +83,7 @@ class FunctionWithValvesModel(BaseModel): class FunctionResponse(BaseModel): + version_id: str | None = None id: str user_id: str | None = None # may be null for legacy/malformed records type: str @@ -96,6 +102,7 @@ class FunctionUserResponse(FunctionResponse): class FunctionForm(BaseModel): + commit_message: str | None = None id: str name: str content: str @@ -107,79 +114,117 @@ class FunctionValves(BaseModel): class FunctionsTable: - async def insert_new_function( + async def _lock_function(self, session, id): + # UPDATE also serializes writers on SQLite, where SELECT FOR UPDATE does not. + await session.execute(update(Function).where(Function.id == id).values(version_id=Function.version_id)) + return await session.get(Function, id, populate_existing=True) + + async def _write_function( self, - user_id: str, - type: str, - form_data: FunctionForm, - db: AsyncSession | None = None, - ) -> FunctionModel | None: - function = FunctionModel( - **{ - **form_data.model_dump(), - 'user_id': user_id, - 'type': type, - 'updated_at': int(time.time()), - 'created_at': int(time.time()), - } - ) + session, + resource, + updated, + user_id=None, + version_id=None, + module=None, + merge_meta=False, + ): + updated = dict(updated) + message = updated.pop('commit_message', None) + updated.pop('version_id', None) # Imported pointers never belong to this resource. + before = function_snapshot(resource) + if version_id: + entry = ( + await session.execute(select(FunctionHistory).filter_by(id=version_id, function_id=resource.id)) + ).scalar_one_or_none() + if not entry: + raise HTTPException(404, 'Version not found') + # The prepared candidate must be the exact selected saved configuration. + if function_snapshot(updated) != function_snapshot(entry.snapshot): + raise HTTPException(400, 'Version configuration does not match the saved snapshot') + if module is not None: + validate_valves(module, updated.get('valves', resource.valves)) + if merge_meta: + updated['meta'] = {**(resource.meta or {}), **updated.get('meta', {})} + for key, value in updated.items(): + setattr(resource, key, value) + after = function_snapshot(resource) + if version_id: + resource.version_id = version_id + elif after != before or not resource.version_id: + entry = FunctionHistories.new_entry( + resource.id, after, user_id or resource.user_id or '', resource.version_id, message + ) + session.add(entry) + resource.version_id = entry.id + resource.updated_at = int(time.time()) - try: - async with get_async_db_context(db) as db: - result = Function(**function.model_dump()) - db.add(result) - await db.commit() - if result: - return FunctionModel.model_validate(result) - else: - return None - except Exception as e: - log.exception(f'Error creating a new function: {e}') - return None + async def insert_new_function(self, user_id, type, form_data, db=None, module=None): + async with get_async_db_context(db) as session: + try: + function = Function( + **form_data.model_dump(exclude={'commit_message'}), + user_id=user_id, + type=type, + is_active=False, + is_global=False, + updated_at=int(time.time()), + created_at=int(time.time()), + ) + session.add(function) + await self._write_function( + session, + function, + {'commit_message': form_data.commit_message}, + user_id, + module=module, + ) + await session.flush() + result = FunctionModel.model_validate(function) + await session.commit() + return result + except Exception: + await session.rollback() + raise - async def sync_functions( - self, - user_id: str, - functions: list[FunctionWithValvesModel], - db: AsyncSession | None = None, - ) -> list[FunctionWithValvesModel]: - # Synchronize functions by updating existing ones, inserting new ones, - # and removing those that are no longer present. - try: - async with get_async_db_context(db) as db: - # Get existing functions - result = await db.execute(select(Function)) - existing_functions = result.scalars().all() - existing_ids = {func.id for func in existing_functions} - - # Prepare a set of new function IDs - new_function_ids = {func.id for func in functions} - - # Update or insert functions + async def sync_functions(self, user_id, functions, db=None, modules=None): + async with get_async_db_context(db) as session: + try: + # Lock all existing rows in a stable order before applying the batch. + ids = (await session.execute(select(Function.id).order_by(Function.id))).scalars().all() + existing = {id: await self._lock_function(session, id) for id in ids} + incoming = {func.id for func in functions} for func in functions: - func_data = func.model_dump() - func_data['valves'] = encrypt_valves(func_data['valves']) if func_data.get('valves') else None - func_data['user_id'] = user_id - func_data['updated_at'] = int(time.time()) - - if func.id in existing_ids: - await db.execute(update(Function).filter_by(id=func.id).values(**func_data)) - else: - new_func = Function(**func_data) - db.add(new_func) - - # Remove functions that are no longer present - for func in existing_functions: - if func.id not in new_function_ids: - await db.delete(func) - - await db.commit() - - result = await db.execute(select(Function)) - return [FunctionModel.model_validate(func) for func in result.scalars().all()] - except Exception as e: - log.exception(f'Error syncing functions for user {user_id}: {e}') - return [] + data = func.model_dump(exclude={'version_id'}) + data['valves'] = encrypt_valves(data.get('valves')) + data['user_id'] = user_id + resource = existing.get(func.id) + if resource is None: + resource = Function(**data) + session.add(resource) + await self._write_function( + session, + resource, + data, + user_id, + module=(modules or {}).get(func.id), + ) + for id in set(existing) - incoming: + await session.execute(delete(FunctionHistory).filter_by(function_id=id)) + await session.delete(existing[id]) + await session.flush() + rows = (await session.execute(select(Function))).scalars().all() + result = [ + FunctionWithValvesModel.model_validate( + {**FunctionModel.model_validate(row).model_dump(), 'valves': decrypt_valves(row.valves)} + ) + for row in rows + ] + await session.commit() + return result + except Exception: + await session.rollback() + raise async def get_function_by_id(self, id: str, db: AsyncSession | None = None) -> FunctionModel | None: try: @@ -329,27 +374,8 @@ class FunctionsTable: except Exception: return None - async def update_function_metadata_by_id( - self, id: str, metadata: dict, db: AsyncSession | None = None - ) -> FunctionModel | None: - async with get_async_db_context(db) as db: - try: - function = await db.get(Function, id) - - if function: - if function.meta: - function.meta = {**function.meta, **metadata} - else: - function.meta = metadata - - function.updated_at = int(time.time()) - await db.commit() - return FunctionModel.model_validate(function) - else: - return None - except Exception as e: - log.exception(f'Error updating function metadata by id {id}: {e}') - return None + async def update_function_metadata_by_id(self, id, metadata, db=None, user_id=None): + return await self.update_function_by_id(id, {'meta': metadata}, db=db, user_id=user_id, merge_meta=True) async def get_user_valves_by_id_and_user_id( self, id: str, user_id: str, db: AsyncSession | None = None @@ -396,23 +422,29 @@ class FunctionsTable: return None async def update_function_by_id( - self, id: str, updated: dict, db: AsyncSession | None = None - ) -> FunctionModel | None: - async with get_async_db_context(db) as db: + self, id, updated, db=None, user_id=None, version_id=None, module=None, merge_meta=False + ): + async with get_async_db_context(db) as session: try: - await db.execute( - update(Function) - .filter_by(id=id) - .values( - **updated, - updated_at=int(time.time()), - ) + function = await self._lock_function(session, id) + if not function: + raise ValueError('Function not found') + await self._write_function( + session, + function, + updated, + user_id, + version_id, + module, + merge_meta, ) - await db.commit() - function = await db.get(Function, id) - return FunctionModel.model_validate(function) if function else None + await session.flush() + result = FunctionModel.model_validate(function) + await session.commit() + return result except Exception: - return None + await session.rollback() + raise async def deactivate_all_functions(self, db: AsyncSession | None = None) -> bool | None: async with get_async_db_context(db) as db: @@ -431,6 +463,8 @@ class FunctionsTable: async def delete_function_by_id(self, id: str, db: AsyncSession | None = None) -> bool: async with get_async_db_context(db) as db: try: + await self._lock_function(db, id) + await db.execute(delete(FunctionHistory).filter_by(function_id=id)) await db.execute(delete(Function).filter_by(id=id)) await db.commit() diff --git a/backend/open_webui/models/tool_history.py b/backend/open_webui/models/tool_history.py new file mode 100644 index 0000000000..1e42b0b8ea --- /dev/null +++ b/backend/open_webui/models/tool_history.py @@ -0,0 +1,160 @@ +"""Immutable snapshots of tool configuration; the live row remains Production.""" + +import difflib +import time +import uuid +from copy import deepcopy + +from fastapi import HTTPException +from open_webui.internal.db import Base, get_async_db_context +from pydantic import BaseModel, ConfigDict +from sqlalchemy import JSON, BigInteger, Column, Text, select, update + + +class ToolHistory(Base): + __tablename__ = 'tool_history' + id = Column(Text, primary_key=True) + tool_id = Column(Text, nullable=False, index=True) + parent_id = Column(Text, nullable=True) + snapshot = Column(JSON, nullable=False) + user_id = Column(Text, nullable=False) + commit_message = Column(Text, nullable=True) + created_at = Column(BigInteger, nullable=False) + + +class ToolHistoryResponse(BaseModel): + model_config = ConfigDict(from_attributes=True) + id: str + tool_id: str + parent_id: str | None = None + user_id: str + commit_message: str | None = None + created_at: int + user: dict | None = None + + +class ToolHistoryModel(ToolHistoryResponse): + snapshot: dict + + +class ToolHistoryTable: + async def delete_history_entry(self, tool_id, history_id, db=None): + from open_webui.models.tools import Tool + + async with get_async_db_context(db) as session: + try: + # Serialize with production switches on both SQLite and PostgreSQL. + await session.execute(update(Tool).where(Tool.id == tool_id).values(version_id=Tool.version_id)) + model = await session.get(Tool, tool_id, populate_existing=True) + if not model: + return False + if model.version_id == history_id: + raise HTTPException(400, 'Cannot delete the current version') + entry = ( + await session.execute(select(ToolHistory).filter_by(id=history_id, tool_id=tool_id)) + ).scalar_one_or_none() + if not entry: + return False + await session.execute( + update(ToolHistory) + .where(ToolHistory.tool_id == tool_id, ToolHistory.parent_id == history_id) + .values(parent_id=entry.parent_id) + ) + await session.delete(entry) + await session.commit() + return True + except Exception: + await session.rollback() + raise + + def new_entry(self, tool_id, snapshot, user_id, parent_id=None, commit_message=None): + return ToolHistory( + id=str(uuid.uuid4()), + tool_id=tool_id, + snapshot=snapshot, + user_id=user_id, + parent_id=parent_id, + commit_message=commit_message, + created_at=int(time.time()), + ) + + async def get_history_by_id(self, tool_id, history_id, db=None): + from open_webui.models.users import User + + async with get_async_db_context(db) as session: + entry = ( + await session.execute(select(ToolHistory).filter_by(tool_id=tool_id, id=history_id)) + ).scalar_one_or_none() + if not entry: + return None + result = ToolHistoryModel.model_validate(entry) + author = (await session.execute(select(User.name).where(User.id == entry.user_id))).scalar_one_or_none() + result.user = {'name': author} if author else None + return result + + async def get_history_by_tool_id(self, tool_id, page=1, db=None): + from open_webui.models.users import User + + async with get_async_db_context(db) as session: + columns = [getattr(ToolHistory, key) for key in ToolHistoryResponse.model_fields if key != 'user'] + rows = ( + ( + await session.execute( + select(*columns, User.name.label('author_name')) + .outerjoin(User, User.id == ToolHistory.user_id) + .where(ToolHistory.tool_id == tool_id) + .order_by(ToolHistory.created_at.desc(), ToolHistory.id.desc()) + .offset((max(1, page) - 1) * 20) + .limit(20) + ) + ) + .mappings() + .all() + ) + return [ + ToolHistoryResponse( + **{key: value for key, value in row.items() if key != 'author_name'}, + user={'name': row['author_name']} if row['author_name'] else None, + ) + for row in rows + ] + + +ToolHistories = ToolHistoryTable() + + +def tool_snapshot(resource): + data = ( + resource if isinstance(resource, dict) else {key: getattr(resource, key) for key in ('name', 'content', 'meta')} + ) + meta = data.get('meta') or {} + meta = meta.model_dump() if isinstance(meta, BaseModel) else deepcopy(meta) + meta.setdefault('description', None) + if not meta.get('i18n'): + meta.pop('i18n', None) + for key in ('manifest', 'has_user_valves'): + meta.pop(key, None) + return {'name': data.get('name'), 'content': data.get('content') or '', 'meta': meta} + + +def tool_diff(before, after): + left, right = before.snapshot, after.snapshot + metadata = { + key: {'before': left.get(key), 'after': right.get(key)} + for key in ('name', 'meta') + if left.get(key) != right.get(key) + } + old, new = left.get('content') or '', right.get('content') or '' + # splitlines handles a missing final newline and CRLF without breaking the renderer. + patch = '\n'.join( + difflib.unified_diff( + old.splitlines(), new.splitlines(), fromfile='selected.py', tofile='production.py', lineterm='' + ) + ) + return { + 'from_id': before.id, + 'to_id': after.id, + 'metadata': metadata, + 'content_diff': patch, + 'line_endings_only': old != new and not patch, + } diff --git a/backend/open_webui/models/tools.py b/backend/open_webui/models/tools.py index c08c6ce5da..be99c7550e 100644 --- a/backend/open_webui/models/tools.py +++ b/backend/open_webui/models/tools.py @@ -7,10 +7,12 @@ import time # local imports from open_webui.internal.db import Base, JSONField, get_async_db_context -from open_webui.models.access_grants import AccessGrantModel, AccessGrants +from open_webui.models.access_grants import AccessGrant, AccessGrantModel, AccessGrants from open_webui.models.groups import Groups +from open_webui.models.tool_history import ToolHistories, ToolHistory, tool_snapshot from open_webui.models.users import UserResponse, Users -from open_webui.utils.valves import decrypt_valves, encrypt_valves +from open_webui.utils.valves import decrypt_valves, encrypt_valves, validate_valves +from fastapi import HTTPException from pydantic import BaseModel, ConfigDict, Field from sqlalchemy import BigInteger, Column, String, Text, delete, select, update from sqlalchemy.ext.asyncio import AsyncSession @@ -22,6 +24,7 @@ class Tool(Base): # database table definition __tablename__ = 'tool' id = Column(String, primary_key=True, unique=True) + version_id = Column(Text, nullable=True) user_id = Column(String, index=True) # owner user id name = Column(Text) # human-readable label content = Column(Text) # Python source code @@ -34,6 +37,7 @@ class Tool(Base): # database table definition class ToolMeta(BaseModel): + model_config = ConfigDict(extra='allow') i18n: dict[str, dict[str, str]] | None = None description: str | None = None manifest: dict | None = {} @@ -41,6 +45,7 @@ class ToolMeta(BaseModel): class ToolModel(BaseModel): + version_id: str | None = None id: str user_id: str | None = None # may be null for legacy/malformed records name: str @@ -66,6 +71,7 @@ class ToolUserModel(ToolModel): class ToolResponse(BaseModel): + version_id: str | None = None id: str user_id: str | None = None # may be null for legacy/malformed records name: str @@ -86,6 +92,7 @@ class ToolAccessResponse(ToolUserResponse): class ToolForm(BaseModel): + commit_message: str | None = None id: str name: str content: str @@ -98,6 +105,48 @@ class ToolValves(BaseModel): class ToolsTable: + async def _lock_tool(self, session, id): + # UPDATE also serializes writers on SQLite, where SELECT FOR UPDATE does not. + await session.execute(update(Tool).where(Tool.id == id).values(version_id=Tool.version_id)) + return await session.get(Tool, id, populate_existing=True) + + async def _write_tool( + self, + session, + resource, + updated, + user_id=None, + version_id=None, + module=None, + ): + updated = dict(updated) + message = updated.pop('commit_message', None) + updated.pop('version_id', None) # Imported pointers never belong to this resource. + before = tool_snapshot(resource) + if version_id: + entry = ( + await session.execute(select(ToolHistory).filter_by(id=version_id, tool_id=resource.id)) + ).scalar_one_or_none() + if not entry: + raise HTTPException(404, 'Version not found') + # The prepared candidate must be the exact selected saved configuration. + if tool_snapshot(updated) != tool_snapshot(entry.snapshot): + raise HTTPException(400, 'Version configuration does not match the saved snapshot') + if module is not None: + validate_valves(module, updated.get('valves', resource.valves)) + for key, value in updated.items(): + setattr(resource, key, value) + after = tool_snapshot(resource) + if version_id: + resource.version_id = version_id + elif after != before or not resource.version_id: + entry = ToolHistories.new_entry( + resource.id, after, user_id or resource.user_id or '', resource.version_id, message + ) + session.add(entry) + resource.version_id = entry.id + resource.updated_at = int(time.time()) + async def _get_access_grants(self, tool_id: str, db: AsyncSession | None = None) -> list[AccessGrantModel]: return await AccessGrants.get_grants_by_resource('tool', tool_id, db=db) @@ -113,34 +162,29 @@ class ToolsTable: ) return tool_model - async def insert_new_tool( - self, - user_id: str, - form_data: ToolForm, - specs: list[dict], - db: AsyncSession | None = None, - ) -> ToolModel | None: - async with get_async_db_context(db) as db: + async def insert_new_tool(self, user_id, form_data, specs, db=None, module=None): + async with get_async_db_context(db) as session: try: - result = Tool( - **{ - **form_data.model_dump(exclude={'access_grants'}), - 'specs': specs, - 'user_id': user_id, - 'updated_at': int(time.time()), - 'created_at': int(time.time()), - } + data = form_data.model_dump(exclude={'access_grants', 'commit_message'}) + tool = Tool( + **data, specs=specs, user_id=user_id, created_at=int(time.time()), updated_at=int(time.time()) ) - db.add(result) - await db.commit() - await AccessGrants.set_access_grants('tool', result.id, form_data.access_grants, db=db) - if result: - return await self._to_tool_model(result, db=db) - else: - return None - except Exception as e: - log.exception(f'Error creating a new tool: {e}') - return None # creation failed + session.add(tool) + await self._write_tool( + session, + tool, + {'commit_message': form_data.commit_message}, + user_id, + module=module, + ) + await session.flush() + grants = await AccessGrants.replace_access_grants(session, 'tool', tool.id, form_data.access_grants) + result = await self._to_tool_model(tool, access_grants=grants) + await session.commit() + return result + except Exception: + await session.rollback() + raise async def get_tool_by_id( self, @@ -182,7 +226,16 @@ class ToolsTable: # Skip Tool.content (plugin source, potentially large) via a # column select; Row attributes satisfy from_attributes. stmt = ( - select(Tool.id, Tool.user_id, Tool.name, Tool.specs, Tool.meta, Tool.updated_at, Tool.created_at) + select( + Tool.id, + Tool.version_id, + Tool.user_id, + Tool.name, + Tool.specs, + Tool.meta, + Tool.updated_at, + Tool.created_at, + ) if defer_content else select(Tool) ).order_by(Tool.updated_at.desc()) @@ -311,31 +364,48 @@ class ToolsTable: log.exception(f'Error updating user valves by id {id} and user_id {user_id}: {e}') return None - async def update_tool_by_id(self, id: str, updated: dict, db: AsyncSession | None = None) -> ToolModel | None: - try: - async with get_async_db_context(db) as db: - access_grants = updated.pop('access_grants', None) - await db.execute(update(Tool).filter_by(id=id).values(**updated, updated_at=int(time.time()))) - await db.commit() - if access_grants is not None: - await AccessGrants.set_access_grants('tool', id, access_grants, db=db) - - # populate_existing: the Core update above bypasses any identity-map copy - tool = await db.get(Tool, id, populate_existing=True) - return await self._to_tool_model(tool, db=db) - except Exception: - return None - - async def delete_tool_by_id(self, id: str, db: AsyncSession | None = None) -> bool: - try: - async with get_async_db_context(db) as db: - await AccessGrants.revoke_all_access('tool', id, db=db) - await db.execute(delete(Tool).filter_by(id=id)) - await db.commit() + async def update_tool_by_id( + self, id, updated, db=None, user_id=None, version_id=None, module=None, allow_code_changes=True + ): + async with get_async_db_context(db) as session: + try: + tool = await self._lock_tool(session, id) + if not tool: + raise ValueError('Tool not found') + if not allow_code_changes and updated.get('content', tool.content) != tool.content: + raise HTTPException(401, 'You do not have permission to change executable Tool code') + updated = dict(updated) + grants = updated.pop('access_grants', None) + await self._write_tool(session, tool, updated, user_id, version_id, module) + if grants is not None: + await AccessGrants.replace_access_grants(session, 'tool', id, grants) + await session.flush() + grants = ( + (await session.execute(select(AccessGrant).filter_by(resource_type='tool', resource_id=id))) + .scalars() + .all() + ) + result = await self._to_tool_model( + tool, access_grants=[AccessGrantModel.model_validate(g) for g in grants] + ) + await session.commit() + return result + except Exception: + await session.rollback() + raise + async def delete_tool_by_id(self, id, db=None): + async with get_async_db_context(db) as session: + try: + await self._lock_tool(session, id) + await session.execute(delete(AccessGrant).filter_by(resource_type='tool', resource_id=id)) + await session.execute(delete(ToolHistory).filter_by(tool_id=id)) + await session.execute(delete(Tool).filter_by(id=id)) + await session.commit() return True - except Exception: - return False + except Exception: + await session.rollback() + raise Tools = ToolsTable() # singleton tool registry diff --git a/backend/open_webui/routers/functions.py b/backend/open_webui/routers/functions.py index 18fa4ed433..e5ceedcd66 100644 --- a/backend/open_webui/routers/functions.py +++ b/backend/open_webui/routers/functions.py @@ -13,6 +13,7 @@ from open_webui.constants import ERROR_MESSAGES from open_webui.env import AIOHTTP_CLIENT_SESSION_SSL, AIOHTTP_CLIENT_TIMEOUT, ENABLE_FUNCTIONS from open_webui.events import EVENTS, build_event, dispatch_event_functions, publish_event, schedule_webhook_dispatch from open_webui.internal.db import get_async_session +from open_webui.models.function_history import FunctionHistories, function_diff from open_webui.models.functions import ( FunctionForm, FunctionModel, @@ -29,6 +30,7 @@ from open_webui.utils.plugin import ( load_function_module_by_id, replace_imports, resolve_valves_schema_options, + set_function_module_in_cache, ) from pydantic import BaseModel, HttpUrl from sqlalchemy.ext.asyncio import AsyncSession @@ -168,22 +170,27 @@ async def sync_functions( db: AsyncSession = Depends(get_async_session), ): try: + modules = {} + source_modules = {} + previous_ids = {entry.id for entry in await Functions.get_functions(db=db)} for function in form_data.functions: function.content = replace_imports(function.content) - function_module, function_type, frontmatter = await load_function_module_by_id( - function.id, - content=function.content, + module, function.type, frontmatter, source_module = await load_function_module_by_id( + function.id, content=function.content ) - - if hasattr(function_module, 'Valves') and function.valves: - Valves = function_module.Valves - try: - Valves(**{k: v for k, v in function.valves.items() if v is not None}) - except Exception as e: - log.exception(f'Error validating valves for function {function.id}: {e}') - raise e - - return await Functions.sync_functions(user.id, form_data.functions, db=db) + function.meta.manifest = frontmatter + function.meta.toggle = function.type == 'filter' and bool(getattr(module, 'toggle', False)) + modules[function.id] = module + source_modules[function.id] = source_module + result = await Functions.sync_functions(user.id, form_data.functions, db=db, modules=modules) + for function in result: + set_function_module_in_cache( + request, function.id, function.content, modules[function.id], source_modules[function.id] + ) + for id in previous_ids - {entry.id for entry in result}: + get_functions_cache(request).pop(id, None) + get_function_contents_cache(request).pop(id, None) + return result except Exception as e: log.exception(f'Failed to load a function: {e}') raise HTTPException( @@ -216,24 +223,22 @@ async def create_new_function( if function is None: try: form_data.content = replace_imports(form_data.content) - function_module, function_type, frontmatter = await load_function_module_by_id( + function_module, function_type, frontmatter, source_module = await load_function_module_by_id( form_data.id, content=form_data.content, ) form_data.meta.manifest = frontmatter + form_data.meta.toggle = function_type == 'filter' and bool(getattr(function_module, 'toggle', False)) - FUNCTIONS = get_functions_cache(request) - FUNCTIONS[form_data.id] = function_module - - function = await Functions.insert_new_function(user.id, function_type, form_data, db=db) + function = await Functions.insert_new_function( + user.id, function_type, form_data, db=db, module=function_module + ) function_cache_dir = CACHE_DIR / 'functions' / form_data.id function_cache_dir.mkdir(parents=True, exist_ok=True) - if function_type == 'filter' and getattr(function_module, 'toggle', None): - await Functions.update_function_metadata_by_id(form_data.id, {'toggle': True}, db=db) - if function: + set_function_module_in_cache(request, function.id, function.content, function_module, source_module) await publish_event( request, EVENTS.FUNCTION_CREATED, @@ -384,23 +389,28 @@ async def update_function_by_id( user=Depends(get_admin_user), db: AsyncSession = Depends(get_async_session), ): + return await _update_function(request, id, form_data, user, db) + + +async def _update_function(request, id, form_data, user, db, version_id=None): try: - form_data.content = replace_imports(form_data.content) - function_module, function_type, frontmatter = await load_function_module_by_id(id, content=form_data.content) + if version_id is None: + form_data.content = replace_imports(form_data.content) + function_module, function_type, frontmatter, source_module = await load_function_module_by_id( + id, content=form_data.content + ) form_data.meta.manifest = frontmatter - FUNCTIONS = get_functions_cache(request) - FUNCTIONS[id] = function_module + form_data.meta.toggle = function_type == 'filter' and bool(getattr(function_module, 'toggle', False)) updated = {**form_data.model_dump(exclude={'id'}), 'type': function_type} - log.debug(updated) - function = await Functions.update_function_by_id(id, updated, db=db) - - if function_type == 'filter' and getattr(function_module, 'toggle', None): - await Functions.update_function_metadata_by_id(id, {'toggle': True}, db=db) + function = await Functions.update_function_by_id( + id, updated, db=db, user_id=user.id, version_id=version_id, module=function_module + ) if function: + set_function_module_in_cache(request, function.id, function.content, function_module, source_module) await publish_event( request, EVENTS.FUNCTION_UPDATED, @@ -420,7 +430,7 @@ async def update_function_by_id( except Exception as e: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, - detail=ERROR_MESSAGES.DEFAULT(e, 'Error updating function'), + detail=str(e), ) @@ -666,3 +676,74 @@ async def update_function_user_valves_by_id( status_code=status.HTTP_401_UNAUTHORIZED, detail=ERROR_MESSAGES.NOT_FOUND, ) + + +async def require_function_history_access(id, user, db): + resource = await Functions.get_function_by_id(id, db=db) + if not resource: + raise HTTPException(404, 'Not found') + return resource + + +async def require_function_history_entry(id, history_id, db): + entry = await FunctionHistories.get_history_by_id(id, history_id, db=db) + if not entry: + raise HTTPException(404, 'Version not found') + return entry + + +@router.get('/id/{id}/history') +async def get_function_history( + id: str, page: int = 1, user=Depends(get_admin_user), db: AsyncSession = Depends(get_async_session) +): + await require_function_history_access(id, user, db) + return await FunctionHistories.get_history_by_function_id(id, page, db=db) + + +@router.get('/id/{id}/history/diff') +async def get_function_history_diff( + id: str, from_id: str, to_id: str, user=Depends(get_admin_user), db: AsyncSession = Depends(get_async_session) +): + await require_function_history_access(id, user, db) + before = await require_function_history_entry(id, from_id, db) + after = await require_function_history_entry(id, to_id, db) + return function_diff(before, after) + + +@router.get('/id/{id}/history/{history_id}') +async def get_function_history_entry( + id: str, history_id: str, user=Depends(get_admin_user), db: AsyncSession = Depends(get_async_session) +): + await require_function_history_access(id, user, db) + return await require_function_history_entry(id, history_id, db) + + +@router.delete('/id/{id}/history/{history_id}') +async def delete_function_history_entry( + id: str, history_id: str, user=Depends(get_admin_user), db: AsyncSession = Depends(get_async_session) +): + await require_function_history_access(id, user, db) + if not await FunctionHistories.delete_history_entry(id, history_id, db=db): + raise HTTPException(404, 'Version not found') + return True + + +class FunctionVersionForm(BaseModel): + version_id: str + + +@router.post('/id/{id}/update/version', response_model=FunctionModel) +async def set_function_production( + request: Request, + id: str, + form_data: FunctionVersionForm, + user=Depends(get_admin_user), + db: AsyncSession = Depends(get_async_session), +): + await require_function_history_access(id, user, db) + entry = await require_function_history_entry(id, form_data.version_id, db) + try: + saved = FunctionForm(id=id, **entry.snapshot) + except ValueError as error: + raise HTTPException(400, str(error)) from error + return await _update_function(request, id, saved, user, db, version_id=entry.id) diff --git a/backend/open_webui/routers/tools.py b/backend/open_webui/routers/tools.py index d5bf49dfef..383d6669b1 100644 --- a/backend/open_webui/routers/tools.py +++ b/backend/open_webui/routers/tools.py @@ -23,6 +23,7 @@ from open_webui.models.access_grants import AccessGrants from open_webui.models.config import Config from open_webui.models.groups import Groups from open_webui.models.oauth_sessions import OAuthSessions +from open_webui.models.tool_history import ToolHistories, tool_diff from open_webui.models.tools import ( ToolAccessResponse, ToolForm, @@ -44,6 +45,7 @@ from open_webui.utils.plugin import ( load_tool_module_by_id, replace_imports, resolve_valves_schema_options, + set_tool_module_in_cache, ) from open_webui.utils.tools import connect_mcp_server, get_tool_servers from open_webui.utils.tools import get_tool_specs as get_local_tool_specs @@ -422,20 +424,20 @@ async def create_new_tools( ) form_data.content = replace_imports(form_data.content) - tool_module, frontmatter = await load_tool_module_by_id(form_data.id, content=form_data.content) + tool_module, frontmatter, source_module = await load_tool_module_by_id( + form_data.id, content=form_data.content + ) form_data.meta.manifest = frontmatter form_data.meta.has_user_valves = hasattr(tool_module, 'UserValves') - TOOLS = get_tools_cache(request) - TOOLS[form_data.id] = tool_module - - specs = get_local_tool_specs(TOOLS[form_data.id]) - tools = await Tools.insert_new_tool(user.id, form_data, specs, db=db) + specs = get_local_tool_specs(tool_module) + tools = await Tools.insert_new_tool(user.id, form_data, specs, db=db, module=tool_module) tool_cache_dir = CACHE_DIR / 'tools' / form_data.id tool_cache_dir.mkdir(parents=True, exist_ok=True) if tools: + set_tool_module_in_cache(request, tools.id, tools.content, tool_module, source_module) await publish_event( request, EVENTS.TOOL_CREATED, @@ -526,6 +528,10 @@ async def update_tools_by_id( user=Depends(get_verified_user), db: AsyncSession = Depends(get_async_session), ): + return await _update_tool(request, id, form_data, user, db) + + +async def _update_tool(request, id, form_data, user, db, version_id=None): """Update an existing tool's source code and metadata.""" tools = await Tools.get_tool_by_id(id, db=db) if not tools: @@ -551,45 +557,49 @@ async def update_tools_by_id( detail=ERROR_MESSAGES.UNAUTHORIZED, ) - # Content edits trigger exec on load — gate them behind workspace.tools (matches /create). - if form_data.content != tools.content: - if user.role != 'admin' and not ( - await has_permission(user.id, 'workspace.tools', await Config.get('user.permissions'), db=db) - or await has_permission(user.id, 'workspace.tools_import', await Config.get('user.permissions'), db=db) - ): - raise HTTPException( - status_code=status.HTTP_401_UNAUTHORIZED, - detail=ERROR_MESSAGES.UNAUTHORIZED, - ) + # Check again under the row lock when committing, in case Production changes meanwhile. + allow_code_changes = await can_change_tool_code(user, db) + if form_data.content != tools.content and not allow_code_changes: + raise HTTPException(401, ERROR_MESSAGES.UNAUTHORIZED) try: - form_data.content = replace_imports(form_data.content) - tool_module, frontmatter = await load_tool_module_by_id(id, content=form_data.content) + if version_id is None: + form_data.content = replace_imports(form_data.content) + tool_module, frontmatter, source_module = await load_tool_module_by_id(id, content=form_data.content) form_data.meta.manifest = frontmatter form_data.meta.has_user_valves = hasattr(tool_module, 'UserValves') - TOOLS = get_tools_cache(request) - TOOLS[id] = tool_module + specs = get_local_tool_specs(tool_module) - specs = get_local_tool_specs(TOOLS[id]) - - form_data.access_grants = await filter_allowed_access_grants( - await Config.get('user.permissions'), - user.id, - user.role, - form_data.access_grants, - 'sharing.public_tools', - ) + if version_id is None: + form_data.access_grants = await filter_allowed_access_grants( + await Config.get('user.permissions'), + user.id, + user.role, + form_data.access_grants, + 'sharing.public_tools', + ) updated = { **form_data.model_dump(exclude={'id'}), 'specs': specs, } - log.debug(updated) - tools = await Tools.update_tool_by_id(id, updated, db=db) + if version_id is not None: + updated.pop('access_grants', None) + + tools = await Tools.update_tool_by_id( + id, + updated, + db=db, + user_id=user.id, + version_id=version_id, + module=tool_module, + allow_code_changes=allow_code_changes, + ) if tools: + set_tool_module_in_cache(request, tools.id, tools.content, tool_module, source_module) await publish_event( request, EVENTS.TOOL_UPDATED, @@ -609,7 +619,7 @@ async def update_tools_by_id( except Exception as e: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, - detail=ERROR_MESSAGES.DEFAULT(e, 'Error updating tool'), + detail=str(e), ) @@ -1022,3 +1032,89 @@ async def update_tools_user_valves_by_id( status_code=status.HTTP_401_UNAUTHORIZED, detail=ERROR_MESSAGES.NOT_FOUND, ) + + +async def can_change_tool_code(user, db): + return user.role == 'admin' or ( + await has_permission(user.id, 'workspace.tools', await Config.get('user.permissions'), db=db) + or await has_permission(user.id, 'workspace.tools_import', await Config.get('user.permissions'), db=db) + ) + + +async def require_tool_history_access(id, user, db): + resource = await Tools.get_tool_by_id(id, db=db) + if not resource: + raise HTTPException(404, 'Not found') + if not ( + (user.role == 'admin' and BYPASS_ADMIN_ACCESS_CONTROL) + or resource.user_id == user.id + or await AccessGrants.has_access( + user_id=user.id, resource_type='tool', resource_id=id, permission='write', db=db + ) + ): + raise HTTPException(401, ERROR_MESSAGES.ACCESS_PROHIBITED) + return resource + + +async def require_tool_history_entry(id, history_id, db): + entry = await ToolHistories.get_history_by_id(id, history_id, db=db) + if not entry: + raise HTTPException(404, 'Version not found') + return entry + + +@router.get('/id/{id}/history') +async def get_tool_history( + id: str, page: int = 1, user=Depends(get_verified_user), db: AsyncSession = Depends(get_async_session) +): + await require_tool_history_access(id, user, db) + return await ToolHistories.get_history_by_tool_id(id, page, db=db) + + +@router.get('/id/{id}/history/diff') +async def get_tool_history_diff( + id: str, from_id: str, to_id: str, user=Depends(get_verified_user), db: AsyncSession = Depends(get_async_session) +): + await require_tool_history_access(id, user, db) + before = await require_tool_history_entry(id, from_id, db) + after = await require_tool_history_entry(id, to_id, db) + return tool_diff(before, after) + + +@router.get('/id/{id}/history/{history_id}') +async def get_tool_history_entry( + id: str, history_id: str, user=Depends(get_verified_user), db: AsyncSession = Depends(get_async_session) +): + await require_tool_history_access(id, user, db) + return await require_tool_history_entry(id, history_id, db) + + +@router.delete('/id/{id}/history/{history_id}') +async def delete_tool_history_entry( + id: str, history_id: str, user=Depends(get_verified_user), db: AsyncSession = Depends(get_async_session) +): + await require_tool_history_access(id, user, db) + if not await ToolHistories.delete_history_entry(id, history_id, db=db): + raise HTTPException(404, 'Version not found') + return True + + +class ToolVersionForm(BaseModel): + version_id: str + + +@router.post('/id/{id}/update/version', response_model=ToolModel) +async def set_tool_production( + request: Request, + id: str, + form_data: ToolVersionForm, + user=Depends(get_verified_user), + db: AsyncSession = Depends(get_async_session), +): + await require_tool_history_access(id, user, db) + entry = await require_tool_history_entry(id, form_data.version_id, db) + try: + saved = ToolForm(id=id, **entry.snapshot) + except ValueError as error: + raise HTTPException(400, str(error)) from error + return await _update_tool(request, id, saved, user, db, version_id=entry.id) diff --git a/backend/open_webui/utils/plugin.py b/backend/open_webui/utils/plugin.py index cc9dc27b47..281c22375f 100644 --- a/backend/open_webui/utils/plugin.py +++ b/backend/open_webui/utils/plugin.py @@ -8,6 +8,7 @@ import subprocess import sys import tempfile import types +import uuid from importlib import util from typing import Any @@ -205,6 +206,7 @@ def replace_imports(content): # May the intent of the one who wrote it survive every # import and transformation, as a deed survives the generations. async def load_tool_module_by_id(tool_id, content=None): + """Prepare a Tool instance, frontmatter, and Python module without changing the live cache.""" if not ENABLE_TOOLS: raise RuntimeError('Tools are disabled by ENABLE_PLUGINS or ENABLE_TOOLS') @@ -215,9 +217,6 @@ async def load_tool_module_by_id(tool_id, content=None): raise Exception(f'Toolkit not found: {tool_id}') content = tool.content - - content = replace_imports(content) - await Tools.update_tool_by_id(tool_id, {'content': content}) else: frontmatter = extract_frontmatter(content) # Install required packages found within the frontmatter. @@ -225,7 +224,8 @@ async def load_tool_module_by_id(tool_id, content=None): # offload to a thread so it doesn't block the event loop. await asyncio.to_thread(install_frontmatter_requirements, frontmatter.get('requirements', '')) - module_name = f'tool_{tool_id}' + content = replace_imports(content) + module_name = f'tool_{tool_id}_{uuid.uuid4().hex}' module = types.ModuleType(module_name) sys.modules[module_name] = module @@ -246,18 +246,19 @@ async def load_tool_module_by_id(tool_id, content=None): # Create and return the object if the class 'Tools' is found in the module if hasattr(module, 'Tools'): - return module.Tools(), frontmatter + return module.Tools(), frontmatter, module else: raise Exception('No Tools class found in the module') except Exception as e: log.error(f'Error loading module: {tool_id}: {e}') - del sys.modules[module_name] # Clean up raise e finally: os.unlink(temp_file.name) + sys.modules.pop(module_name, None) async def load_function_module_by_id(function_id: str, content: str | None = None): + """Prepare a Function instance, type, frontmatter, and Python module without publishing it.""" if not ENABLE_FUNCTIONS: raise RuntimeError('Functions are disabled by ENABLE_PLUGINS or ENABLE_FUNCTIONS') @@ -267,15 +268,13 @@ async def load_function_module_by_id(function_id: str, content: str | None = Non if not function: raise Exception(f'Function not found: {function_id}') content = function.content - - content = replace_imports(content) - await Functions.update_function_by_id(function_id, {'content': content}) else: frontmatter = extract_frontmatter(content) # `pip install` via subprocess can block for a long time; offload it. await asyncio.to_thread(install_frontmatter_requirements, frontmatter.get('requirements', '')) - module_name = f'function_{function_id}' + content = replace_imports(content) + module_name = f'function_{function_id}_{uuid.uuid4().hex}' module = types.ModuleType(module_name) sys.modules[module_name] = module @@ -296,24 +295,21 @@ async def load_function_module_by_id(function_id: str, content: str | None = Non # Create appropriate object based on available class type in the module if hasattr(module, 'Pipe'): - return module.Pipe(), 'pipe', frontmatter + return module.Pipe(), 'pipe', frontmatter, module elif hasattr(module, 'Filter'): - return module.Filter(), 'filter', frontmatter + return module.Filter(), 'filter', frontmatter, module elif hasattr(module, 'Action'): - return module.Action(), 'action', frontmatter + return module.Action(), 'action', frontmatter, module elif hasattr(module, 'Event'): - return module.Event(), 'event', frontmatter + return module.Event(), 'event', frontmatter, module else: raise Exception('No Function class found in the module') except Exception as e: log.error(f'Error loading module: {function_id}: {e}') - # Cleanup by removing the module in case of error - del sys.modules[module_name] - - await Functions.update_function_by_id(function_id, {'is_active': False}) raise e finally: os.unlink(temp_file.name) + sys.modules.pop(module_name, None) def _state_cache(request, name: str) -> dict: @@ -338,6 +334,26 @@ def get_function_contents_cache(request) -> dict: return _state_cache(request, 'FUNCTION_CONTENTS') +def set_tool_module_in_cache(request, tool_id, content, tool_module, source_module): + previous = sys.modules.get(f'tool_{tool_id}') + if previous is not None: + sys.modules.pop(previous.__name__, None) + sys.modules[source_module.__name__] = source_module + sys.modules[f'tool_{tool_id}'] = source_module + get_tools_cache(request)[tool_id] = tool_module + get_tool_contents_cache(request)[tool_id] = content + + +def set_function_module_in_cache(request, function_id, content, function_module, source_module): + previous = sys.modules.get(f'function_{function_id}') + if previous is not None: + sys.modules.pop(previous.__name__, None) + sys.modules[source_module.__name__] = source_module + sys.modules[f'function_{function_id}'] = source_module + get_functions_cache(request)[function_id] = function_module + get_function_contents_cache(request)[function_id] = content + + async def get_tool_module_from_cache(request, tool_id, load_from_db=True): if not ENABLE_TOOLS: raise RuntimeError('Tools are disabled by ENABLE_PLUGINS or ENABLE_TOOLS') @@ -353,25 +369,16 @@ async def get_tool_module_from_cache(request, tool_id, load_from_db=True): raise Exception(f'Tool not found: {tool_id}') content = tool.content - new_content = replace_imports(content) - if new_content != content: - content = new_content - # Update the tool content in the database - await Tools.update_tool_by_id(tool_id, {'content': content}) - if tool_id in tool_contents_cache and tool_id in tools_cache: if tool_contents_cache[tool_id] == content: return tools_cache[tool_id], None - tool_module, frontmatter = await load_tool_module_by_id(tool_id, content) else: if tool_id in tools_cache: return tools_cache[tool_id], None - tool_module, frontmatter = await load_tool_module_by_id(tool_id) - - tools_cache[tool_id] = tool_module - tool_contents_cache[tool_id] = content + tool_module, frontmatter, source_module = await load_tool_module_by_id(tool_id, content) + set_tool_module_in_cache(request, tool_id, content, tool_module, source_module) return tool_module, frontmatter @@ -397,17 +404,10 @@ async def get_function_module_from_cache( raise Exception(f'Function not found: {function_id}') content = function.content - new_content = replace_imports(content) - if new_content != content: - content = new_content - # Update the function content in the database - await Functions.update_function_by_id(function_id, {'content': content}) - if function_id in function_contents_cache and function_id in functions_cache: if function_contents_cache[function_id] == content: return functions_cache[function_id], None, None - function_module, function_type, frontmatter = await load_function_module_by_id(function_id, content) else: # Load from cache (e.g. "stream" hook) # This is useful for performance reasons @@ -415,10 +415,15 @@ async def get_function_module_from_cache( if function_id in functions_cache: return functions_cache[function_id], None, None - function_module, function_type, frontmatter = await load_function_module_by_id(function_id) + try: + function_module, function_type, frontmatter, source_module = await load_function_module_by_id( + function_id, content + ) + except Exception: + await Functions.update_function_by_id(function_id, {'is_active': False}) + raise - functions_cache[function_id] = function_module - function_contents_cache[function_id] = content + set_function_module_in_cache(request, function_id, content, function_module, source_module) return function_module, function_type, frontmatter diff --git a/backend/open_webui/utils/tools.py b/backend/open_webui/utils/tools.py index e3622d09ef..f80ff77d2b 100644 --- a/backend/open_webui/utils/tools.py +++ b/backend/open_webui/utils/tools.py @@ -116,7 +116,12 @@ from open_webui.utils.json_codec import JSONCodec from open_webui.utils.mcp.client import MCPClient, OAuthTokenAuth from open_webui.utils.misc import is_string_allowed from open_webui.utils.oauth import get_system_oauth_token -from open_webui.utils.plugin import get_tool_contents_cache, get_tools_cache, load_tool_module_by_id +from open_webui.utils.plugin import ( + get_tool_contents_cache, + get_tools_cache, + load_tool_module_by_id, + set_tool_module_in_cache, +) from open_webui.utils.terminals import ( TERMINAL_CONTEXT_HEADER, get_terminal_server_url, @@ -375,9 +380,8 @@ async def get_tools(request: Request, tool_ids: list[str], user: UserModel, extr tool_contents_cache = get_tool_contents_cache(request) module = tools_cache.get(tool_id) if module is None or tool_contents_cache.get(tool_id) != tool.content: - module, _ = await load_tool_module_by_id(tool_id, content=tool.content) - tools_cache[tool_id] = module - tool_contents_cache[tool_id] = tool.content + module, _, source_module = await load_tool_module_by_id(tool_id, content=tool.content) + set_tool_module_in_cache(request, tool_id, tool.content, module, source_module) __user__ = { **extra_params['__user__'], diff --git a/backend/open_webui/utils/valves.py b/backend/open_webui/utils/valves.py index 338f272bd7..e21945dd24 100644 --- a/backend/open_webui/utils/valves.py +++ b/backend/open_webui/utils/valves.py @@ -4,6 +4,8 @@ import logging from functools import lru_cache from cryptography.fernet import Fernet, InvalidToken +from fastapi import HTTPException +from pydantic import ValidationError from open_webui.env import ENABLE_VALVE_ENCRYPTION, WEBUI_SECRET_KEY from open_webui.utils.json_codec import JSONCodec @@ -39,3 +41,20 @@ def decrypt_valves(valves) -> dict: return {} return decrypted if isinstance(decrypted, dict) else {} + + +def validate_valves(module, valves): + if hasattr(module, 'Valves'): + values = decrypt_valves(valves) or {} + try: + # Validate without replacing the stored values (including unused keys/secrets). + module.Valves(**{key: value for key, value in values.items() if value is not None}) + except Exception as error: + detail = ( + '; '.join( + f'{".".join(map(str, item["loc"]))}: {item["msg"]}' for item in error.errors(include_input=False) + ) + if isinstance(error, ValidationError) + else str(error) + ) + raise HTTPException(400, f'Current Valves are incompatible with this code: {detail}') from error diff --git a/src/lib/apis/functions/index.ts b/src/lib/apis/functions/index.ts index 47346b4a20..198214191d 100644 --- a/src/lib/apis/functions/index.ts +++ b/src/lib/apis/functions/index.ts @@ -518,3 +518,101 @@ export const updateUserValvesById = async (token: string, id: string, valves: ob return res; }; + +export type FunctionHistoryEntry = { + id: string; + function_id: string; + parent_id: string | null; + user_id: string; + commit_message: string | null; + created_at: number; + user?: { name: string } | null; +}; + +export type FunctionSnapshot = { + name: string; + content: string; + meta: Record; +}; + +export const getFunctionHistory = async ( + token: string, + id: string, + page = 1 +): Promise => { + const response = await fetch( + `${WEBUI_API_BASE_URL}/functions/id/${encodeURIComponent(id)}/history?page=${page}`, + { + method: 'GET', + headers: { authorization: `Bearer ${token}` } + } + ); + if (!response.ok) throw await response.json(); + return response.json(); +}; + +export const getFunctionHistoryEntry = async ( + token: string, + id: string, + versionId: string +): Promise => { + const response = await fetch( + `${WEBUI_API_BASE_URL}/functions/id/${encodeURIComponent(id)}/history/${encodeURIComponent(versionId)}`, + { + method: 'GET', + headers: { authorization: `Bearer ${token}` } + } + ); + if (!response.ok) throw await response.json(); + return response.json(); +}; + +export const getFunctionHistoryDiff = async ( + token: string, + id: string, + fromId: string, + toId: string +) => { + const response = await fetch( + `${WEBUI_API_BASE_URL}/functions/id/${encodeURIComponent(id)}/history/diff?${new URLSearchParams({ from_id: fromId, to_id: toId })}`, + { + method: 'GET', + headers: { authorization: `Bearer ${token}` } + } + ); + if (!response.ok) throw await response.json(); + return response.json(); +}; + +export const deleteFunctionHistoryVersion = async ( + token: string, + id: string, + versionId: string +) => { + const response = await fetch( + `${WEBUI_API_BASE_URL}/functions/id/${encodeURIComponent(id)}/history/${encodeURIComponent(versionId)}`, + { + method: 'DELETE', + headers: { authorization: `Bearer ${token}` } + } + ); + if (!response.ok) throw await response.json(); + return response.json(); +}; + +export const setProductionFunctionVersion = async ( + token: string, + id: string, + versionId: string +) => { + const response = await fetch( + `${WEBUI_API_BASE_URL}/functions/id/${encodeURIComponent(id)}/update/version`, + { + method: 'POST', + headers: { authorization: `Bearer ${token}`, 'Content-Type': 'application/json' }, + body: JSON.stringify({ version_id: versionId }) + } + ); + if (!response.ok) throw await response.json(); + return response.json(); +}; diff --git a/src/lib/apis/tools/index.ts b/src/lib/apis/tools/index.ts index 4f21efa5d9..b3cdbcb876 100644 --- a/src/lib/apis/tools/index.ts +++ b/src/lib/apis/tools/index.ts @@ -496,3 +496,93 @@ export const updateUserValvesById = async (token: string, id: string, valves: ob return res; }; + +export type ToolHistoryEntry = { + id: string; + tool_id: string; + parent_id: string | null; + user_id: string; + commit_message: string | null; + created_at: number; + user?: { name: string } | null; +}; + +export type ToolSnapshot = { + name: string; + content: string; + meta: Record; +}; + +export const getToolHistory = async ( + token: string, + id: string, + page = 1 +): Promise => { + const response = await fetch( + `${WEBUI_API_BASE_URL}/tools/id/${encodeURIComponent(id)}/history?page=${page}`, + { + method: 'GET', + headers: { authorization: `Bearer ${token}` } + } + ); + if (!response.ok) throw await response.json(); + return response.json(); +}; + +export const getToolHistoryEntry = async ( + token: string, + id: string, + versionId: string +): Promise => { + const response = await fetch( + `${WEBUI_API_BASE_URL}/tools/id/${encodeURIComponent(id)}/history/${encodeURIComponent(versionId)}`, + { + method: 'GET', + headers: { authorization: `Bearer ${token}` } + } + ); + if (!response.ok) throw await response.json(); + return response.json(); +}; + +export const getToolHistoryDiff = async ( + token: string, + id: string, + fromId: string, + toId: string +) => { + const response = await fetch( + `${WEBUI_API_BASE_URL}/tools/id/${encodeURIComponent(id)}/history/diff?${new URLSearchParams({ from_id: fromId, to_id: toId })}`, + { + method: 'GET', + headers: { authorization: `Bearer ${token}` } + } + ); + if (!response.ok) throw await response.json(); + return response.json(); +}; + +export const deleteToolHistoryVersion = async (token: string, id: string, versionId: string) => { + const response = await fetch( + `${WEBUI_API_BASE_URL}/tools/id/${encodeURIComponent(id)}/history/${encodeURIComponent(versionId)}`, + { + method: 'DELETE', + headers: { authorization: `Bearer ${token}` } + } + ); + if (!response.ok) throw await response.json(); + return response.json(); +}; + +export const setProductionToolVersion = async (token: string, id: string, versionId: string) => { + const response = await fetch( + `${WEBUI_API_BASE_URL}/tools/id/${encodeURIComponent(id)}/update/version`, + { + method: 'POST', + headers: { authorization: `Bearer ${token}`, 'Content-Type': 'application/json' }, + body: JSON.stringify({ version_id: versionId }) + } + ); + if (!response.ok) throw await response.json(); + return response.json(); +}; diff --git a/src/lib/components/admin/Functions/FunctionEditor.svelte b/src/lib/components/admin/Functions/FunctionEditor.svelte index e891da510c..d720616079 100644 --- a/src/lib/components/admin/Functions/FunctionEditor.svelte +++ b/src/lib/components/admin/Functions/FunctionEditor.svelte @@ -1,8 +1,12 @@ - + + + + + +
+ + +
+ + selectVersion(resource.version_id)} + /> + {#if loading} +
+ {:else if error} + + {:else} + {#if history.some((entry) => entry.id !== resource.version_id)}
{/if} + {#each history.filter((entry) => entry.id !== resource.version_id) as entry (entry.id)} + selectVersion(entry.id)} + onDelete={() => { + deleteVersionId = entry.id; + show = false; + showDeleteVersion = true; + }} + /> + {/each} + {#if page > 1 || history.length === 20} +
+ + +
+ {/if} + {/if} + +
+ +
diff --git a/src/lib/components/common/CodeEditor.svelte b/src/lib/components/common/CodeEditor.svelte index 81ee22ef2d..c3521adeb2 100644 --- a/src/lib/components/common/CodeEditor.svelte +++ b/src/lib/components/common/CodeEditor.svelte @@ -27,13 +27,14 @@ export let boilerplate = ''; export let value = ''; export let className = 'text-sm'; + export let readOnly = false; export let onSave = () => {}; - export let onChange = () => {}; + export let onChange: (value: string) => void = () => {}; let _value = ''; - $: if (value) { + $: if (value !== undefined) { updateValue(); } @@ -89,9 +90,20 @@ let editorTheme = new Compartment(); let editorLanguage = new Compartment(); const editorPlaceholder = new Compartment(); + const editorReadOnly = new Compartment(); $: if (codeEditor) { codeEditor.dispatch({ - effects: editorPlaceholder.reconfigure(placeholder($i18n.t('Enter your code here...'))) + effects: editorReadOnly.reconfigure([ + EditorState.readOnly.of(readOnly), + EditorView.editable.of(!readOnly) + ]) + }); + } + $: if (codeEditor) { + codeEditor.dispatch({ + effects: editorPlaceholder.reconfigure( + readOnly ? [] : placeholder($i18n.t('Enter your code here...')) + ) }); } @@ -185,6 +197,7 @@ print("${endTag}") }; export const formatPythonCodeHandler = async () => { + if (readOnly) return false; if (codeEditor) { const res = await ( $user?.role === 'admin' @@ -194,14 +207,14 @@ print("${endTag}") toast.error(`${error}`); return null; }); - if (res && res.code) { + if (!readOnly && res && typeof res.code === 'string') { const formattedCode = res.code; codeEditor.dispatch({ changes: [{ from: 0, to: codeEditor.state.doc.length, insert: formattedCode }] }); _value = formattedCode; - onChange(_value); + if (!readOnly) onChange(_value); await tick(); toast.success($i18n.t('Code formatted successfully')); @@ -214,13 +227,15 @@ print("${endTag}") let extensions = [ basicSetup, + EditorView.contentAttributes.of({ tabindex: '0' }), keymap.of([{ key: 'Tab', run: acceptCompletion }, indentWithTab]), indentUnit.of(' '), editorPlaceholder.of([]), + editorReadOnly.of([EditorState.readOnly.of(readOnly), EditorView.editable.of(!readOnly)]), EditorView.updateListener.of((e) => { if (e.docChanged) { _value = e.state.doc.toString(); - onChange(_value); + if (!readOnly) onChange(_value); } }), editorTheme.of([]), @@ -241,7 +256,7 @@ print("${endTag}") }; onMount(() => { - if (value === '') { + if (value === '' && !readOnly) { value = boilerplate; } @@ -293,14 +308,15 @@ print("${endTag}") }); const keydownHandler = async (e) => { + if (!codeEditor?.hasFocus) return; if ((e.ctrlKey || e.metaKey) && e.key === 's') { e.preventDefault(); - onSave(); + if (!readOnly) onSave(); } // Format code when Ctrl + Shift + F is pressed - if ((e.ctrlKey || e.metaKey) && e.shiftKey && e.key === 'f') { + if ((e.ctrlKey || e.metaKey) && e.shiftKey && e.key.toLowerCase() === 'f') { e.preventDefault(); await formatPythonCodeHandler(); } diff --git a/src/lib/components/common/TranslationTable.svelte b/src/lib/components/common/TranslationTable.svelte index 2028662777..30ad5d1a52 100644 --- a/src/lib/components/common/TranslationTable.svelte +++ b/src/lib/components/common/TranslationTable.svelte @@ -1,4 +1,7 @@ + + + + + +
+ + +
+ + selectVersion(resource.version_id)} + /> + {#if loading} +
+ {:else if error} + + {:else} + {#if history.some((entry) => entry.id !== resource.version_id)}
{/if} + {#each history.filter((entry) => entry.id !== resource.version_id) as entry (entry.id)} + selectVersion(entry.id)} + onDelete={() => { + deleteVersionId = entry.id; + show = false; + showDeleteVersion = true; + }} + /> + {/each} + {#if page > 1 || history.length === 20} +
+ + +
+ {/if} + {/if} + +
+ +
diff --git a/src/lib/components/workspace/Tools/ToolkitEditor.svelte b/src/lib/components/workspace/Tools/ToolkitEditor.svelte index 66e820fb6f..ab04832c61 100644 --- a/src/lib/components/workspace/Tools/ToolkitEditor.svelte +++ b/src/lib/components/workspace/Tools/ToolkitEditor.svelte @@ -1,10 +1,13 @@ -