refac

backend/open_webui/utils/automations.py

@@ -7,6 +7,7 @@ Follows the utils/<feature>.py pattern (cf. utils/channels.py, utils/task.py).
 The scheduler_worker_loop handles all time-based background work:
   - Automation execution (claim_due → execute)
   - Calendar event alerts (upcoming events → socket + webhook notifications)
+  - One-shot chat timers
 
 Environment:
     SCHEDULER_POLL_INTERVAL             – seconds between polls (default: 10)
@@ -190,8 +191,18 @@ async def scheduler_worker_loop(app) -> None:
     SCHEDULER_POLL_INTERVAL env var (default: 10 seconds).
     """
     log.info(f'Scheduler worker started (poll interval: {SCHEDULER_POLL_INTERVAL}s)')
+
     while True:
         try:
+            # ── Timers ──
+            try:
+                from open_webui.utils.timers import claim_due_timers, execute_due_timer
+
+                for timer_id, claim_id in await claim_due_timers(int(time.time_ns()), limit=10):
+                    asyncio.create_task(execute_due_timer(app, timer_id, claim_id))
+            except Exception:
+                log.exception('Scheduler: timer error')
+
             # ── Automations ──
             if await Config.get('automations.enable'):
                 try:

backend/open_webui/utils/subagents.py

@@ -67,7 +67,7 @@ def _build_request(source: Request, user_id: str, *, internal: bool) -> Request:
     return request
 
 
-async def process_pending_subagent_results(
+async def process_pending_internal_messages(
     source_request: Request,
     parent_chat_id: str,
     user_id: str,
@@ -92,40 +92,74 @@ async def process_pending_subagent_results(
             message
             for message in messages.values()
             if message.get('role') == 'user'
-            and (message.get('meta') or {}).get('async_subagent_result') is True
             and not message.get('childrenIds')
+            and (
+                (message.get('meta') or {}).get('async_subagent_result') is True
+                or (
+                    (message.get('meta') or {}).get('internal') is True
+                    and (message.get('meta') or {}).get('type') == 'timer'
+                )
+            )
         ]
         if not pending:
             return
 
         first = pending[0]
+        first_meta = first.get('meta') or {}
+        kind = 'subagent' if first_meta.get('async_subagent_result') is True else 'timer'
         parent_id = first.get('parentId')
+        if kind == 'timer' and first_meta.get('timer_id'):
+            timer = await Chats.get_chat_by_id(first_meta['timer_id'])
+            run = {**run, **(((timer.meta or {}).get('run') if timer else None) or {})}
         model_id = first.get('model') or run['model_id']
-        batch = [
-            message
-            for message in pending
-            if message.get('parentId') == parent_id and (message.get('model') or model_id) == model_id
-        ]
+        if kind == 'timer':
+            batch = [
+                message
+                for message in pending
+                if message.get('parentId') == parent_id
+                and (message.get('model') or model_id) == model_id
+                and (message.get('meta') or {}).get('internal') is True
+                and (message.get('meta') or {}).get('type') == 'timer'
+            ]
+        else:
+            batch = [
+                message
+                for message in pending
+                if message.get('parentId') == parent_id
+                and (message.get('model') or model_id) == model_id
+                and (message.get('meta') or {}).get('async_subagent_result') is True
+            ]
         combined_content = '\n\n'.join(message.get('content', '') for message in batch if message.get('content'))
-        delegation_ids = [
-            message['meta']['delegation_id'] for message in batch if (message.get('meta') or {}).get('delegation_id')
-        ]
-        subagent_chat_ids = [
-            message['meta']['subagent_chat_id']
-            for message in batch
-            if (message.get('meta') or {}).get('subagent_chat_id')
-        ]
-        combined_meta = {'async_subagent_result': True}
-        if len(delegation_ids) == 1:
-            combined_meta['delegation_id'] = delegation_ids[0]
-        elif delegation_ids:
-            combined_meta['delegation_ids'] = delegation_ids
-        if len(subagent_chat_ids) == 1:
-            combined_meta['subagent_chat_id'] = subagent_chat_ids[0]
-        elif subagent_chat_ids:
-            combined_meta['subagent_chat_ids'] = subagent_chat_ids
+        if kind == 'timer':
+            timer_ids = [message['meta']['timer_id'] for message in batch if (message.get('meta') or {}).get('timer_id')]
+            combined_meta = {'internal': True, 'type': 'timer'}
+            if len(timer_ids) == 1:
+                combined_meta['timer_id'] = timer_ids[0]
+            elif timer_ids:
+                combined_meta['timer_ids'] = timer_ids
+        else:
+            delegation_ids = [
+                message['meta']['delegation_id']
+                for message in batch
+                if (message.get('meta') or {}).get('delegation_id')
+            ]
+            subagent_chat_ids = [
+                message['meta']['subagent_chat_id']
+                for message in batch
+                if (message.get('meta') or {}).get('subagent_chat_id')
+            ]
+            combined_meta = {'async_subagent_result': True}
+            if len(delegation_ids) == 1:
+                combined_meta['delegation_id'] = delegation_ids[0]
+            elif delegation_ids:
+                combined_meta['delegation_ids'] = delegation_ids
+            if len(subagent_chat_ids) == 1:
+                combined_meta['subagent_chat_id'] = subagent_chat_ids[0]
+            elif subagent_chat_ids:
+                combined_meta['subagent_chat_ids'] = subagent_chat_ids
 
-        reuse_message = len(batch) == 1 and not (first.get('meta') or {}).get('async_subagent_pending')
+        pending_flag = 'timer_pending' if kind == 'timer' else 'async_subagent_pending'
+        reuse_message = len(batch) == 1 and not (first.get('meta') or {}).get(pending_flag)
         user_message_id = first['id'] if reuse_message else str(uuid4())
         if not reuse_message:
             removed_ids = {message['id'] for message in batch}
@@ -211,12 +245,13 @@ async def process_pending_subagent_results(
             'id': assistant_message_id,
             'parent_id': parent_id,
             'user_message': user_message,
-            'session_id': run.get('session_id') or f'subagent-result:{parent_chat_id}',
+            'session_id': run.get('session_id') or f'{kind}-result:{parent_chat_id}',
             'background_tasks': {},
             'tool_ids': run.get('tool_ids') or [],
             'skill_ids': run.get('skill_ids') or [],
             'filter_ids': run.get('filter_ids') or [],
             'features': run.get('features') or {},
+            'files': run.get('files') or [],
             'variables': run.get('variables') or {},
         }
         if run.get('terminal_id'):
@@ -582,7 +617,7 @@ async def delegate(
                 room=f'user:{user.id}',
             )
         if not await has_active_tasks(request.app.state.redis, parent_chat_id):
-            await process_pending_subagent_results(request, parent_chat_id, user.id, run)
+            await process_pending_internal_messages(request, parent_chat_id, user.id, run)
         if cancelled:
             raise asyncio.CancelledError
         return result

backend/open_webui/utils/timers.py

@@ -0,0 +1,355 @@
+"""Durable one-shot timers backed by internal child chats."""
+
+from __future__ import annotations
+
+import asyncio
+import copy
+import json
+import logging
+import re
+import time
+from datetime import datetime, timezone
+from typing import Literal
+from uuid import uuid4
+
+from fastapi import Request
+from sqlalchemy import select
+from starlette.datastructures import Headers
+
+from open_webui.internal.db import get_async_db
+from open_webui.models.chat_messages import ChatMessages
+from open_webui.models.chats import Chat, ChatForm, Chats
+from open_webui.models.users import UserModel, Users
+from open_webui.tasks import has_active_tasks
+
+log = logging.getLogger(__name__)
+
+_RELATIVE_TIME = re.compile(r"^(?:\+|in\s+)?(\d+)\s*(s|sec(?:onds?)?|m|min(?:utes?)?|h|hours?|d|days?)$")
+_RFC3339_TIME = re.compile(r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d+)?(?:Z|[+-]\d{2}:\d{2})$")
+_TIME_UNITS_NS = {
+    's': 1_000_000_000,
+    'm': 60 * 1_000_000_000,
+    'h': 60 * 60 * 1_000_000_000,
+    'd': 24 * 60 * 60 * 1_000_000_000,
+}
+_timer_locks: dict[str, asyncio.Lock] = {}
+
+
+def parse_timer_at(value: str) -> int:
+    """Normalize a relative offset or timezone-aware RFC 3339 timestamp."""
+    raw = value.strip()
+    now = time.time_ns()
+    relative = _RELATIVE_TIME.fullmatch(raw.lower())
+    if relative:
+        count = int(relative.group(1))
+        if count <= 0:
+            raise ValueError('at must be in the future.')
+        return now + count * _TIME_UNITS_NS[relative.group(2)[0]]
+
+    if not _RFC3339_TIME.fullmatch(raw):
+        raise ValueError(
+            'at must be a relative time such as 10s or in 10 seconds, '
+            'or an RFC 3339 timestamp with a timezone.'
+        )
+    try:
+        parsed = datetime.fromisoformat(raw.replace('Z', '+00:00'))
+    except ValueError as exc:
+        raise ValueError(
+            'at must be a relative time such as 10s or in 10 seconds, '
+            'or an RFC 3339 timestamp with a timezone.'
+        ) from exc
+    if parsed.tzinfo is None or parsed.utcoffset() is None:
+        raise ValueError('absolute at values must include an explicit timezone.')
+
+    due_at = int(parsed.timestamp() * 1_000_000_000)
+    if due_at <= now:
+        raise ValueError('at must be in the future.')
+    return due_at
+
+
+async def create_timer(
+    *,
+    prompt: str,
+    at: str,
+    cancel_on: list[Literal['chat.read', 'chat.user_message']] | None,
+    request: Request,
+    user_data: dict,
+    metadata: dict,
+    parent_chat_id: str,
+    parent_message_id: str | None,
+) -> str:
+    prompt = prompt.strip()
+    if not prompt:
+        return 'Error: prompt must not be empty.'
+    if not parent_chat_id or not user_data.get('id'):
+        return 'Error: chat and user context are required.'
+
+    try:
+        due_at = parse_timer_at(at)
+    except ValueError as exc:
+        return f'Error: {exc}'
+
+    selected_events = cancel_on or []
+    allowed_events = {'chat.read', 'chat.user_message'}
+    if any(event not in allowed_events for event in selected_events):
+        return 'Error: cancel_on accepts only chat.read and chat.user_message.'
+    selected_events = list(dict.fromkeys(selected_events))
+
+    model_id = metadata.get('model_id') or (metadata.get('model') or {}).get('id')
+    if not model_id:
+        return 'Error: model context is required.'
+    if metadata.get('direct'):
+        return 'Error: timers are unavailable for direct connections.'
+
+    chat_id = str(uuid4())
+    user_message_id = str(uuid4())
+    user = UserModel(**user_data)
+    run = {
+        'model_id': model_id,
+        'session_id': metadata.get('session_id'),
+        'tool_ids': copy.deepcopy(metadata.get('tool_ids') or []),
+        'skill_ids': copy.deepcopy(metadata.get('skill_ids') or []),
+        'system_prompt': metadata.get('system_prompt'),
+        'filter_ids': copy.deepcopy(metadata.get('filter_ids') or []),
+        'terminal_id': metadata.get('terminal_id'),
+        'features': copy.deepcopy(metadata.get('features') or {}),
+        'files': copy.deepcopy(metadata.get('files') or []),
+        'variables': copy.deepcopy(metadata.get('variables') or {}),
+    }
+
+    chat = await Chats.insert_new_chat(
+        chat_id,
+        user.id,
+        ChatForm(
+            chat={
+                'id': chat_id,
+                'title': f'Timer: {prompt[:60]}',
+                'models': [model_id],
+                'history': {
+                    'currentId': user_message_id,
+                    'messages': {
+                        user_message_id: {
+                            'id': user_message_id,
+                            'parentId': None,
+                            'childrenIds': [],
+                            'role': 'user',
+                            'content': prompt,
+                            'timestamp': int(time.time()),
+                            'models': [model_id],
+                        },
+                    },
+                },
+                'messages': [{'role': 'user', 'content': prompt}],
+            }
+        ),
+        internal_meta={
+            'internal': True,
+            'type': 'timer',
+            'parent_chat_id': parent_chat_id,
+            'parent_message_id': parent_message_id,
+            'timer_at': due_at,
+            'timer_status': 'pending',
+            'timer_model_id': model_id,
+            'timer_task_message_id': user_message_id,
+            'cancel_on': selected_events,
+            'run': run,
+        },
+    )
+    if not chat:
+        return 'Error: failed to create timer.'
+
+    return json.dumps(
+        {
+            'status': 'set',
+            'at': datetime.fromtimestamp(due_at / 1_000_000_000, timezone.utc).isoformat().replace('+00:00', 'Z'),
+            'cancel_on': selected_events,
+        },
+        ensure_ascii=False,
+    )
+
+
+async def claim_due_timers(now_ns: int, limit: int = 10) -> list[tuple[str, str]]:
+    """Claim due timers by moving them from pending to running."""
+    async with get_async_db() as db:
+        stmt = (
+            select(Chat)
+            .where(Chat.meta['internal'].as_boolean().is_(True))
+            .where(Chat.meta['type'].as_string() == 'timer')
+            .where(Chat.meta['timer_status'].as_string() == 'pending')
+        )
+        if db.bind.dialect.name == 'postgresql':
+            stmt = stmt.with_for_update(skip_locked=True)
+
+        result = await db.execute(stmt)
+        rows = [
+            row
+            for row in result.scalars().all()
+            if int((row.meta or {}).get('timer_at') or 0) <= now_ns
+        ]
+        rows.sort(key=lambda row: int((row.meta or {}).get('timer_at') or 0))
+        rows = rows[:limit]
+
+        claimed = []
+        for row in rows:
+            claim_id = str(uuid4())
+            row.meta = {
+                **(row.meta or {}),
+                'timer_status': 'running',
+                'timer_started_at': now_ns,
+                'timer_claim_id': claim_id,
+            }
+            row.updated_at = int(time.time())
+            claimed.append((row.id, claim_id))
+        await db.commit()
+        return claimed
+
+
+async def cancel_timers_for_chat(parent_chat_id: str, event: Literal['chat.read', 'chat.user_message']) -> None:
+    async with get_async_db() as db:
+        result = await db.execute(
+            select(Chat)
+            .where(Chat.meta['internal'].as_boolean().is_(True))
+            .where(Chat.meta['type'].as_string() == 'timer')
+            .where(Chat.meta['parent_chat_id'].as_string() == parent_chat_id)
+            .where(Chat.meta['timer_status'].as_string() == 'pending')
+        )
+        now_ns = int(time.time_ns())
+        for row in result.scalars().all():
+            meta = row.meta or {}
+            if event not in (meta.get('cancel_on') or []):
+                continue
+            row.meta = {
+                **meta,
+                'timer_status': 'cancelled',
+                'timer_cancelled_at': now_ns,
+                'timer_cancelled_by': event,
+            }
+            row.updated_at = int(time.time())
+        await db.commit()
+
+
+async def execute_due_timer(app, timer_id: str, claim_id: str | None = None) -> None:
+    lock = _timer_locks.setdefault(timer_id, asyncio.Lock())
+    async with lock:
+        from open_webui.socket.main import sio
+        from open_webui.utils.subagents import _parent_locks, process_pending_internal_messages
+
+        timer = await Chats.get_chat_by_id(timer_id)
+        if not timer:
+            return
+        meta = timer.meta or {}
+        if meta.get('timer_status') != 'running':
+            return
+        if claim_id is not None and meta.get('timer_claim_id') != claim_id:
+            return
+
+        parent_chat_id = meta.get('parent_chat_id') or ''
+        parent = await Chats.get_chat_by_id_and_user_id(parent_chat_id, timer.user_id)
+        if not parent:
+            await _set_timer_status(timer_id, 'error', timer_error='parent chat no longer exists')
+            return
+
+        prompt_message_id = meta.get('timer_task_message_id')
+        prompt_message = await Chats.get_message_by_id_and_message_id(timer_id, prompt_message_id)
+        if not prompt_message:
+            await _set_timer_status(timer_id, 'error', timer_error='timer task message is missing')
+            return
+
+        user = await Users.get_user_by_id(timer.user_id)
+        if not user:
+            await _set_timer_status(timer_id, 'error', timer_error='timer user no longer exists')
+            return
+
+        run = meta.get('run') or {}
+        model_id = run.get('model_id') or meta.get('timer_model_id')
+        if not model_id:
+            await _set_timer_status(timer_id, 'error', timer_error='model context is missing')
+            return
+
+        prompt = prompt_message.get('content') or ''
+        if isinstance(prompt, list):
+            prompt = ''.join(
+                str(part.get('text', '')) for part in prompt if isinstance(part, dict) and part.get('type') == 'text'
+            )
+
+        user_message_id = str(uuid4())
+        parent_lock = _parent_locks.setdefault(parent_chat_id, asyncio.Lock())
+        async with parent_lock:
+            parent = await Chats.get_chat_by_id_and_user_id(parent_chat_id, timer.user_id)
+            if not parent:
+                await _set_timer_status(timer_id, 'error', timer_error='parent chat no longer exists')
+                return
+            parent_chat = copy.deepcopy(parent.chat or {})
+            history = parent_chat.setdefault('history', {})
+            messages = history.setdefault('messages', {})
+            done_assistants = [
+                message
+                for message in messages.values()
+                if message.get('role') == 'assistant' and message.get('done') is not False
+            ]
+            parent_id = (
+                max(done_assistants, key=lambda message: message.get('timestamp', 0)).get('id')
+                if done_assistants
+                else meta.get('parent_message_id')
+            )
+            pending_meta = {'internal': True, 'type': 'timer', 'timer_id': timer_id}
+            if await has_active_tasks(app.state.redis, parent_chat_id):
+                pending_meta['timer_pending'] = True
+
+            user_message = {
+                'id': user_message_id,
+                'parentId': parent_id,
+                'childrenIds': [],
+                'role': 'user',
+                'content': prompt,
+                'model': model_id,
+                'meta': pending_meta,
+                'timestamp': int(time.time()),
+            }
+            messages[user_message_id] = user_message
+            if parent_id and parent_id in messages:
+                children = messages[parent_id].setdefault('childrenIds', [])
+                if user_message_id not in children:
+                    children.append(user_message_id)
+            await Chats.update_chat_by_id(parent_chat_id, parent_chat)
+            await ChatMessages.upsert_message(user_message_id, parent_chat_id, timer.user_id, user_message)
+            await _set_timer_status(timer_id, 'dispatched', timer_dispatched_at=int(time.time_ns()))
+
+        if user_message['meta'].get('timer_pending') is True:
+            await sio.emit(
+                'events',
+                {
+                    'chat_id': parent_chat_id,
+                    'message_id': user_message_id,
+                    'data': {'type': 'chat:reload'},
+                },
+                room=f'user:{timer.user_id}',
+            )
+        elif not await has_active_tasks(app.state.redis, parent_chat_id):
+            request = Request(
+                {
+                    'type': 'http',
+                    'asgi': {'version': '3.0', 'spec_version': '2.0'},
+                    'method': 'POST',
+                    'path': '/api/v1/timers/internal',
+                    'query_string': b'',
+                    'headers': Headers({}).raw,
+                    'client': ('127.0.0.1', 0),
+                    'server': ('127.0.0.1', 80),
+                    'scheme': 'http',
+                    'app': app,
+                }
+            )
+            request.state.token = None
+            request.state.enable_api_keys = False
+            await process_pending_internal_messages(request, parent_chat_id, user.id, run)
+
+
+async def _set_timer_status(timer_id: str, status: str, **fields) -> None:
+    async with get_async_db() as db:
+        row = await db.get(Chat, timer_id)
+        if not row:
+            return
+        row.meta = {**(row.meta or {}), 'timer_status': status, **fields}
+        row.updated_at = int(time.time())
+        await db.commit()

backend/open_webui/utils/tools.py

@@ -82,6 +82,7 @@ from open_webui.tools.builtin import (
     search_memories,
     search_notes,
     search_web,
+    timer,
     toggle_automation,
     update_automation,
     update_calendar_event,
@@ -574,7 +575,7 @@ async def get_builtin_tools(
         and getattr(request.state, 'internal', False) is not True
         and getattr(request.state, 'direct', False) is not True
     ):
-        builtin_functions.append(delegate_task)
+        builtin_functions.extend([delegate_task, timer])
 
     # Add memory tools when memory is enabled and the model allows this builtin category.
     if (