Implementation:OpenHands OpenHands MaintenanceTask

Knowledge Sources	OpenHands
Domains	Storage, Background_Jobs, Maintenance
Last Updated	2026-02-11 21:00 GMT

Overview

Abstract base class and ORM model for dynamically dispatched maintenance tasks, provided by the OpenHands enterprise storage layer.

Description

This module defines three interrelated components for the maintenance task system:

MaintenanceTaskProcessor is an abstract base class that defines the interface all maintenance task processors must implement. Concrete processors inherit from this class and provide the actual task execution logic.

MaintenanceTaskStatus is an enum that tracks the lifecycle of a maintenance task through states such as pending, running, completed, and failed.

MaintenanceTask is the SQLAlchemy ORM model that persists task records in the database. It stores the task's status, parameters, and crucially, the processor_type string which identifies which processor class should handle the task.

The key architectural feature is dynamic dispatch via get_processor(), which loads a processor class from the processor_type string at runtime. This allows new maintenance task types to be added by simply creating a new processor class and registering its fully-qualified class name, without modifying the task execution infrastructure. The companion set_processor() method stores the fully-qualified class path of a processor on a task record.

Usage

Use MaintenanceTask to schedule and track background maintenance operations. Create a task record with a processor type string, and the task runner will use get_processor() to dynamically load and execute the appropriate processor. Implement MaintenanceTaskProcessor when defining new types of maintenance operations.

Code Reference

Source Location

• Repository: OpenHands
• File: enterprise/storage/maintenance_task.py
• Lines: 1-109

Signature

class MaintenanceTaskProcessor(ABC):
    @abstractmethod
    def process(self, task: 'MaintenanceTask') -> None:
        ...

class MaintenanceTaskStatus(enum.Enum):
    PENDING = "pending"
    RUNNING = "running"
    COMPLETED = "completed"
    FAILED = "failed"

class MaintenanceTask(Base):
    __tablename__ = "maintenance_tasks"

    id: Mapped[int]
    processor_type: Mapped[str]
    status: Mapped[MaintenanceTaskStatus]
    params: Mapped[dict]
    created_at: Mapped[datetime]
    updated_at: Mapped[datetime]

    def get_processor(self) -> MaintenanceTaskProcessor:
        """Dynamically loads and instantiates the processor class from processor_type string."""
        ...

    def set_processor(self, processor: MaintenanceTaskProcessor) -> None:
        """Stores the fully-qualified class path of the processor on this task."""
        ...

Import

from enterprise.storage.maintenance_task import MaintenanceTask, MaintenanceTaskProcessor

I/O Contract

Inputs

MaintenanceTask ORM Fields

Name	Type	Required	Description
processor_type	str	Yes	Fully-qualified class path of the processor to handle this task (e.g., "enterprise.tasks.cleanup.CleanupProcessor")
status	MaintenanceTaskStatus	Yes	Current lifecycle state of the task
params	dict	No	JSON-serializable parameters passed to the processor at execution time

get_processor()

Name	Type	Required	Description
(none)	--	--	Uses the task's own processor_type field to resolve the class

set_processor()

Name	Type	Required	Description
processor	MaintenanceTaskProcessor	Yes	An instance of a processor whose class path will be stored

Outputs

Method	Return Type	Description
get_processor()	MaintenanceTaskProcessor	A new instance of the dynamically loaded processor class
set_processor()	None	Stores the processor's fully-qualified class name in processor_type

MaintenanceTaskStatus Values

Value	Description
PENDING	Task is queued and awaiting execution
RUNNING	Task is currently being processed
COMPLETED	Task finished successfully
FAILED	Task encountered an error during processing

Usage Examples

Defining a Custom Processor

from enterprise.storage.maintenance_task import MaintenanceTask, MaintenanceTaskProcessor

class CleanupProcessor(MaintenanceTaskProcessor):
    def process(self, task: MaintenanceTask) -> None:
        days = task.params.get("retention_days", 30)
        # Perform cleanup logic...

Creating and Dispatching a Task

from enterprise.storage.maintenance_task import MaintenanceTask, MaintenanceTaskStatus

# Create a new maintenance task
task = MaintenanceTask(
    status=MaintenanceTaskStatus.PENDING,
    params={"retention_days": 90}
)

# Set the processor using an instance (stores the class path)
task.set_processor(CleanupProcessor())

# Later, in the task runner, dynamically load and execute:
processor = task.get_processor()
task.status = MaintenanceTaskStatus.RUNNING
processor.process(task)
task.status = MaintenanceTaskStatus.COMPLETED

Related Pages

Environment

• Environment:OpenHands_OpenHands_SaaS_Server_Environment