tiphys.scheduler.models
Models for task scheduling.
Defines the types and structures for scheduled agent tasks.
1""" 2Models for task scheduling. 3 4Defines the types and structures for scheduled agent tasks. 5""" 6 7from datetime import datetime 8from enum import Enum 9from typing import Any 10 11from pydantic import BaseModel, Field 12 13 14class ScheduleType(str, Enum): 15 """Types of schedules supported.""" 16 17 CRON = "cron" # "0 9 * * 1" = Mondays 9am 18 INTERVAL = "interval" # seconds 19 ONCE = "once" # ISO timestamp 20 21 22class TaskStatus(str, Enum): 23 """Status of a scheduled task.""" 24 25 ACTIVE = "active" 26 PAUSED = "paused" 27 COMPLETED = "completed" 28 FAILED = "failed" 29 30 31class ScheduledTask(BaseModel): 32 """A task scheduled to run on a regular basis.""" 33 34 id: str 35 name: str 36 group: str # Scope of the task (e.g. "main", or a channel ID) 37 prompt: str # The prompt to send to the agent 38 schedule_type: ScheduleType 39 schedule_value: str # cron expression, seconds (string), or ISO timestamp 40 status: TaskStatus = TaskStatus.ACTIVE 41 42 # Context for the task 43 agent_id: str = "default" 44 session_key: str | None = None 45 46 # State 47 next_run: datetime | None = None 48 last_run: datetime | None = None 49 run_count: int = 0 50 51 # Metadata 52 created_at: datetime = Field(default_factory=datetime.utcnow) 53 updated_at: datetime = Field(default_factory=datetime.utcnow) 54 metadata: dict[str, Any] = Field(default_factory=dict) 55 56 57class TaskRunResult(BaseModel): 58 """Result of a single task execution.""" 59 60 task_id: str 61 run_id: str 62 timestamp: datetime = Field(default_factory=datetime.utcnow) 63 success: bool 64 output: str | None = None 65 error: str | None = None 66 duration_ms: int = 0
class
ScheduleType(builtins.str, enum.Enum):
15class ScheduleType(str, Enum): 16 """Types of schedules supported.""" 17 18 CRON = "cron" # "0 9 * * 1" = Mondays 9am 19 INTERVAL = "interval" # seconds 20 ONCE = "once" # ISO timestamp
Types of schedules supported.
CRON =
<ScheduleType.CRON: 'cron'>
INTERVAL =
<ScheduleType.INTERVAL: 'interval'>
ONCE =
<ScheduleType.ONCE: 'once'>
class
TaskStatus(builtins.str, enum.Enum):
23class TaskStatus(str, Enum): 24 """Status of a scheduled task.""" 25 26 ACTIVE = "active" 27 PAUSED = "paused" 28 COMPLETED = "completed" 29 FAILED = "failed"
Status of a scheduled task.
ACTIVE =
<TaskStatus.ACTIVE: 'active'>
PAUSED =
<TaskStatus.PAUSED: 'paused'>
COMPLETED =
<TaskStatus.COMPLETED: 'completed'>
FAILED =
<TaskStatus.FAILED: 'failed'>
class
ScheduledTask(pydantic.main.BaseModel):
32class ScheduledTask(BaseModel): 33 """A task scheduled to run on a regular basis.""" 34 35 id: str 36 name: str 37 group: str # Scope of the task (e.g. "main", or a channel ID) 38 prompt: str # The prompt to send to the agent 39 schedule_type: ScheduleType 40 schedule_value: str # cron expression, seconds (string), or ISO timestamp 41 status: TaskStatus = TaskStatus.ACTIVE 42 43 # Context for the task 44 agent_id: str = "default" 45 session_key: str | None = None 46 47 # State 48 next_run: datetime | None = None 49 last_run: datetime | None = None 50 run_count: int = 0 51 52 # Metadata 53 created_at: datetime = Field(default_factory=datetime.utcnow) 54 updated_at: datetime = Field(default_factory=datetime.utcnow) 55 metadata: dict[str, Any] = Field(default_factory=dict)
A task scheduled to run on a regular basis.
class
TaskRunResult(pydantic.main.BaseModel):
58class TaskRunResult(BaseModel): 59 """Result of a single task execution.""" 60 61 task_id: str 62 run_id: str 63 timestamp: datetime = Field(default_factory=datetime.utcnow) 64 success: bool 65 output: str | None = None 66 error: str | None = None 67 duration_ms: int = 0
Result of a single task execution.