Scheduler

Task planner and scheduler for long-running, repeating, and cron tasks.

class folderbot.scheduler.TaskScheduler[source]

Bases: object

Manages scheduled task lifecycle and execution via heartbeat polling.

HEARTBEAT_INTERVAL = 1

__init__(task_store, send_message, summarize)[source]

Parameters:

task_store (TaskStore)
send_message (Callable[[int, str], Awaitable[None]])
summarize (Callable[[str, int], Awaitable[str]])

set_folder_tools(folder_tools)[source]

Set the folder tools reference (called after construction).

Return type:: None
Parameters:: folder_tools (FolderTools)

async start()[source]

Initialize scheduler, restore persisted tasks, start heartbeat.

Return type:: None

async shutdown()[source]

Gracefully stop the heartbeat and save all tasks.

Return type:: None

async create_task(plan)[source]

Register a new task. Returns task_id. Heartbeat will execute it.

Return type:: str
Parameters:: plan (TaskPlan)

async execute_immediate(tool_name, tool_input, user_id=0)[source]

Execute a tool immediately through the scheduler (unified execution path).

This allows all tool calls to go through the scheduler for consistent logging, monitoring, and potential safeguards.

Parameters:

tool_name (str) – Name of the tool to execute
tool_input (dict[str, Any]) – Tool input parameters
user_id (int) – User ID for logging

Return type:

ToolResult

Returns:

ToolResult from the tool execution

list_tasks(user_id, status_filter='')[source]

List tasks for a user, optionally filtered by status.

Return type:

list[TaskPlan]

Parameters:

user_id (int)
status_filter (str)

async cancel_task(task_id, user_id)[source]

Cancel a task. Returns True if found and cancelled.

Return type:

bool

Parameters:

task_id (str)
user_id (int)

get_task_results(task_id, user_id)[source]

Get task plan with results. Returns None if not found/unauthorized.

Return type:

TaskPlan | None

Parameters:

task_id (str)
user_id (int)

class folderbot.scheduler.TaskStore[source]

Bases: object

Persists task plans to SQLite.

__init__(db_path)[source]

Parameters:: db_path (Path)

save_task(plan)[source]

Save or update a task plan.

Return type:: None
Parameters:: plan (TaskPlan)

load_active_tasks()[source]

Load all tasks that were pending or running (for restart recovery).

Return type:: list[TaskPlan]

load_user_tasks(user_id)[source]

Load all tasks for a user.

Return type:: list[TaskPlan]
Parameters:: user_id (int)

Core task scheduler engine.

class folderbot.scheduler.scheduler.TaskScheduler[source]

Bases: object

Manages scheduled task lifecycle and execution via heartbeat polling.

HEARTBEAT_INTERVAL = 1

__init__(task_store, send_message, summarize)[source]

Parameters:

task_store (TaskStore)
send_message (Callable[[int, str], Awaitable[None]])
summarize (Callable[[str, int], Awaitable[str]])

set_folder_tools(folder_tools)[source]

Set the folder tools reference (called after construction).

Return type:: None
Parameters:: folder_tools (FolderTools)

async start()[source]

Initialize scheduler, restore persisted tasks, start heartbeat.

Return type:: None

async shutdown()[source]

Gracefully stop the heartbeat and save all tasks.

Return type:: None

async create_task(plan)[source]

Register a new task. Returns task_id. Heartbeat will execute it.

Return type:: str
Parameters:: plan (TaskPlan)

async execute_immediate(tool_name, tool_input, user_id=0)[source]

Execute a tool immediately through the scheduler (unified execution path).

This allows all tool calls to go through the scheduler for consistent logging, monitoring, and potential safeguards.

Parameters:

tool_name (str) – Name of the tool to execute
tool_input (dict[str, Any]) – Tool input parameters
user_id (int) – User ID for logging

Return type:

ToolResult

Returns:

ToolResult from the tool execution

list_tasks(user_id, status_filter='')[source]

List tasks for a user, optionally filtered by status.

Return type:

list[TaskPlan]

Parameters:

user_id (int)
status_filter (str)

async cancel_task(task_id, user_id)[source]

Cancel a task. Returns True if found and cancelled.

Return type:

bool

Parameters:

task_id (str)
user_id (int)

get_task_results(task_id, user_id)[source]

Get task plan with results. Returns None if not found/unauthorized.

Return type:

TaskPlan | None

Parameters:

task_id (str)
user_id (int)

Data models for the task scheduler.

class folderbot.scheduler.models.ScheduleType[source]

Bases: str, Enum

Type of task schedule.

immediate = 'immediate'

once = 'once'

repeating = 'repeating'

cron = 'cron'

time_limited = 'time_limited'

__new__(value)

class folderbot.scheduler.models.TaskStatus[source]

Bases: str, Enum

Current status of a scheduled task.

pending = 'pending'

running = 'running'

completed = 'completed'

cancelled = 'cancelled'

failed = 'failed'

__new__(value)

class folderbot.scheduler.models.TaskStep[source]

Bases: object

A single tool invocation within a task plan.

tool_name: str

tool_input: dict[str, Any]

__init__(tool_name, tool_input)

Parameters:

tool_name (str)
tool_input (dict[str, Any])

Return type:

None

class folderbot.scheduler.models.Schedule[source]

Bases: object

Schedule configuration for a task.

schedule_type: ScheduleType

delay_seconds: int = 0

interval_seconds: int = 0

cron_expression: str = ''

duration_seconds: int = 0

max_iterations: int = 0

__init__(schedule_type, delay_seconds=0, interval_seconds=0, cron_expression='', duration_seconds=0, max_iterations=0)

Parameters:

schedule_type (ScheduleType)
delay_seconds (int)
interval_seconds (int)
cron_expression (str)
duration_seconds (int)
max_iterations (int)

Return type:

None

class folderbot.scheduler.models.TaskResult[source]

Bases: object

Result of a single task execution.

iteration: int

timestamp: str

tool_name: str

tool_input: dict[str, Any]

content: str

is_error: bool

__init__(iteration, timestamp, tool_name, tool_input, content, is_error)

Parameters:

iteration (int)
timestamp (str)
tool_name (str)
tool_input (dict[str, Any])
content (str)
is_error (bool)

Return type:

None

class folderbot.scheduler.models.TaskPlan[source]

Bases: object

A complete task plan created by Claude.

task_id: str

chat_id: int

user_id: int

description: str

steps: tuple[TaskStep, ...]

schedule: Schedule

status: TaskStatus = 'pending'

created_at: str = ''

started_at: str = ''

completed_at: str = ''

next_run_at: str = ''

deadline_at: str = ''

results: tuple[TaskResult, ...] = ()

current_iteration: int = 0

max_results_kept: int = 100

summarize_on_complete: bool = True

progress_interval: int = 1

last_error: str = ''

consecutive_errors: int = 0

max_consecutive_errors: int = 5

__init__(task_id, chat_id, user_id, description, steps, schedule, status=TaskStatus.pending, created_at='', started_at='', completed_at='', next_run_at='', deadline_at='', results=(), current_iteration=0, max_results_kept=100, summarize_on_complete=True, progress_interval=1, last_error='', consecutive_errors=0, max_consecutive_errors=5)

Parameters:

task_id (str)
chat_id (int)
user_id (int)
description (str)
steps (tuple[TaskStep, ...])
schedule (Schedule)
status (TaskStatus)
created_at (str)
started_at (str)
completed_at (str)
next_run_at (str)
deadline_at (str)
results (tuple[TaskResult, ...])
current_iteration (int)
max_results_kept (int)
summarize_on_complete (bool)
progress_interval (int)
last_error (str)
consecutive_errors (int)
max_consecutive_errors (int)

Return type:

None

SQLite persistence for scheduled tasks.

class folderbot.scheduler.persistence.TaskStore[source]

Bases: object

Persists task plans to SQLite.

__init__(db_path)[source]

Parameters:: db_path (Path)

save_task(plan)[source]

Save or update a task plan.

Return type:: None
Parameters:: plan (TaskPlan)

load_active_tasks()[source]

Load all tasks that were pending or running (for restart recovery).

Return type:: list[TaskPlan]

load_user_tasks(user_id)[source]

Load all tasks for a user.

Return type:: list[TaskPlan]
Parameters:: user_id (int)