Classes

UNSCHEDULED_SORT_KEY = Instant.MAX.timestamp(unit='nanosecond') + 1 module-attribute

Placeholder ordering key for a job that has never been scheduled.

One nanosecond past the largest instant whenever can represent, so it sorts after every real next_run on any platform. Derived rather than hardcoded: a word-size constant such as sys.maxsize sits below the nanosecond timestamp domain on a 32-bit build (and below far-future timestamps everywhere), which would silently invert the intended ordering.

CronTrigger

Internal cron-expression trigger backing Daily and Cron from hassette.scheduler.triggers.

Not part of the public API — use Daily or Cron from hassette.scheduler.triggers instead.

Source code in src/hassette/scheduler/classes.py

class CronTrigger:
    """Internal cron-expression trigger backing ``Daily`` and ``Cron`` from ``hassette.scheduler.triggers``.

    Not part of the public API — use ``Daily`` or ``Cron`` from ``hassette.scheduler.triggers`` instead.
    """

    def __init__(self, cron_expression: str, start: ZonedDateTime | None = None):
        self.cron_expression = cron_expression
        self.start = start
        # Validate expression eagerly at construction time
        base = start or date_utils.now()
        croniter(cron_expression, base.to_stdlib(), ret_type=datetime)

    def __eq__(self, other: object) -> bool:
        if not isinstance(other, CronTrigger):
            return NotImplemented
        return self.cron_expression == other.cron_expression

    def __hash__(self) -> int:
        return hash(self.cron_expression)

    def __str__(self) -> str:
        return f"cron:{self.cron_expression}"

    def first_run_time(self, current_time: ZonedDateTime) -> ZonedDateTime:
        """Return the first cron-grid-aligned run time at or after current_time."""
        # Use start as the croniter anchor, but always snap to the cron grid.
        # This finds the first cron-aligned time at or after start (or current_time if no start).
        anchor = self.start or current_time
        reference = self.start if (self.start is not None and self.start > current_time) else current_time
        return self._next_after(anchor, reference)

    def next_run_time(self, previous_run: ZonedDateTime, current_time: ZonedDateTime) -> ZonedDateTime:
        """Return the next cron-grid-aligned run time after previous_run that is later than current_time."""
        return self._next_after(previous_run, current_time)

    def _next_after(self, anchor: ZonedDateTime, current_time: ZonedDateTime) -> ZonedDateTime:
        cron = croniter(self.cron_expression, anchor.to_stdlib(), ret_type=datetime)
        # Normalise current_time to UTC so the ``next_time > current_dt`` comparison below is
        # unambiguous around DST transitions. During fall-back, ``current_time`` may carry
        # fold=0 (pre-transition, CDT) while croniter returns fold=1 wall-clock values (CST).
        # Without UTC normalisation, the CST occurrence appears UTC-earlier than the CDT
        # anchor and the loop skips the ambiguous slot entirely.
        current_dt_utc = current_time.to_stdlib().astimezone(UTC)
        ambiguous_ticks_skipped = 0
        # Bounded iteration — avoids O(N) spin for sub-second crons after long downtime.
        # 10,000 iterations covers ~2.7 hours of per-second crons, which is generous.
        for _ in range(MAX_CRON_ITERATIONS):
            next_time = cron.get_next()
            if next_time.astimezone(UTC) > current_dt_utc:
                result = self._dst_safe_from_dt(next_time)
                # Log when DST disambiguation is actually relevant: either the returned
                # tick itself is ambiguous (fold=1 was meaningful), or we traversed
                # ambiguous ticks on the way here. Logs the dispatched result (post-
                # disambiguation) to avoid printing a wall-clock string that differs from
                # the instant actually scheduled.
                returned_is_ambiguous = self._is_fall_back_ambiguous(next_time)
                if ambiguous_ticks_skipped or returned_is_ambiguous:
                    LOGGER.info(
                        "CronTrigger(%s): DST fall-back disambiguation — %d ambiguous tick(s) "
                        "traversed; returning %s (fold=1, post-transition)",
                        self.cron_expression,
                        ambiguous_ticks_skipped + (1 if returned_is_ambiguous else 0),
                        result,
                    )
                return result
            # Only count ticks we actually skipped past — the tick that exits the loop is
            # handled above.
            if self._is_fall_back_ambiguous(next_time):
                ambiguous_ticks_skipped += 1
        # Too many iterations — skip ahead from current time.
        # Re-anchor croniter in the *original* timezone so cron expressions like
        # "0 9 * * *" still fire at 09:00 local time, not 09:00 UTC.
        LOGGER.warning(
            "CronTrigger(%s) exceeded %d iterations catching up, skipping ahead from current_time",
            self.cron_expression,
            MAX_CRON_ITERATIONS,
        )
        skip_ahead_dt = current_time.to_stdlib()
        cron = croniter(self.cron_expression, skip_ahead_dt, ret_type=datetime)
        next_time = cron.get_next()
        return self._dst_safe_from_dt(next_time)

    @staticmethod
    def _is_fall_back_ambiguous(dt: datetime) -> bool:
        """Return True when ``dt``'s wall-clock time occurs twice (fall-back ambiguity)."""
        return dt.replace(fold=0).astimezone(UTC) != dt.replace(fold=1).astimezone(UTC)

    @staticmethod
    def _dst_safe_from_dt(dt: datetime) -> ZonedDateTime:
        """Convert a croniter-produced datetime to ZonedDateTime with DST disambiguation.

        Uses ``fold=1`` (post-transition / "later" occurrence) to handle:
        - **Fall-back (fold/repeated time):** prefers the second (post-transition) occurrence.
          The caller is responsible for logging when ambiguity is encountered (see
          ``_next_after``), which emits a single summary per call instead of per-tick spam.
        - **Spring-forward (gap/skipped time):** croniter already advances past the gap, so
          ``fold=1`` and ``fold=0`` produce the same result for non-gap times.

        Args:
            dt: A timezone-aware datetime produced by ``croniter.get_next()`` with a
                ``ZoneInfo``-backed tzinfo.

        Returns:
            A ``ZonedDateTime`` with DST disambiguation applied.
        """
        return ZonedDateTime(dt.replace(fold=1))

first_run_time(current_time: ZonedDateTime) -> ZonedDateTime

Return the first cron-grid-aligned run time at or after current_time.

Source code in src/hassette/scheduler/classes.py

def first_run_time(self, current_time: ZonedDateTime) -> ZonedDateTime:
    """Return the first cron-grid-aligned run time at or after current_time."""
    # Use start as the croniter anchor, but always snap to the cron grid.
    # This finds the first cron-aligned time at or after start (or current_time if no start).
    anchor = self.start or current_time
    reference = self.start if (self.start is not None and self.start > current_time) else current_time
    return self._next_after(anchor, reference)

next_run_time(previous_run: ZonedDateTime, current_time: ZonedDateTime) -> ZonedDateTime

Return the next cron-grid-aligned run time after previous_run that is later than current_time.

Source code in src/hassette/scheduler/classes.py

def next_run_time(self, previous_run: ZonedDateTime, current_time: ZonedDateTime) -> ZonedDateTime:
    """Return the next cron-grid-aligned run time after previous_run that is later than current_time."""
    return self._next_after(previous_run, current_time)

Job dataclass

A live scheduler registration — automatically triggered, entity-waiting, completed, or manual-only.

Source code in src/hassette/scheduler/classes.py

@dataclass(order=True)
class Job:
    """A live scheduler registration — automatically triggered, entity-waiting, completed, or manual-only."""

    sort_index: tuple[int, int] = field(init=False, repr=False)
    """Tuple of (next_run timestamp with nanoseconds, object id) for ordering in a priority queue.

    Assigned the real ordering key when the job transitions to ``SCHEDULED`` with a concrete
    ``next_run`` (see ``set_next_run``/``transition_to``). A job constructed as ``WAITING``,
    ``COMPLETED``, or ``MANUAL`` gets a placeholder key from ``__post_init__`` instead; such a
    job must never be inserted into the heap, so that placeholder is never read for ordering.
    """

    owner_id: str = field(compare=False)
    """Unique string identifier for the owner of the job, e.g., a component or integration name."""

    next_run: ZonedDateTime | None = field(compare=False)
    """Unjittered logical fire time — used as `previous_run` in subsequent trigger calls.

    ``None`` for every status other than ``SCHEDULED`` (waiting, completed, or manual-only).
    """

    fire_at: ZonedDateTime | None = field(default=None, init=False, compare=False)
    """Actual dispatch time, including any jitter offset.

    Equals ``next_run`` when no jitter is configured. Set by ``set_next_run()``, which rolls
    the jitter offset at the same moment it sets ``next_run``, so the value is final before
    the job is enqueued or persisted.
    The pop loop in ``_ScheduledJobQueue.pop_due_and_peek_next`` compares against
    ``fire_at`` (not ``next_run``) to decide when to dispatch. ``None`` for every status
    other than ``SCHEDULED``.
    """

    job: "JobCallable" = field(compare=False)
    """The callable to execute when the job runs."""

    app_key: str = field(default="", compare=False)
    """Configuration-level app key for DB registration (e.g., 'my_app'). Empty for non-App owners."""

    instance_index: int = field(default=0, compare=False)
    """App instance index for DB registration. 0 for non-App owners."""

    instance_name: str | None = field(default=None, compare=False)
    """App instance name, precomputed at registration time from the owning app's app_config.instance_name.
    None for framework-tier jobs and non-App owners."""

    trigger: "TriggerProtocol | None" = field(compare=False, default=None)
    """The trigger that determines the job's schedule. ``None`` for a manual-only job."""

    schedule_status: ScheduleStatus = field(default=ScheduleStatus.SCHEDULED, compare=False)
    """Current schedule status. Defaults to ``SCHEDULED`` so existing direct ``Job(...)``
    constructions (mostly in tests) remain valid without passing it explicitly, as long as
    ``next_run`` is also given — ``__post_init__`` raises ``ValueError`` for the
    ``SCHEDULED``-with-no-``next_run`` combination, mirroring ``transition_to()``'s guard.
    A ``WAITING``/``COMPLETED``/``MANUAL`` construction must pass ``schedule_status``
    explicitly. Real jobs built through ``Scheduler.schedule()``/``Scheduler.register()`` set
    it explicitly, and every transition after construction routes through ``transition_to()``."""

    schedule_status_reason: ScheduleStatusReason | None = field(default=None, compare=False)
    """Optional diagnostic reason qualifying ``schedule_status``. ``None`` for a clean status
    with no override. Set (or cleared) only through ``transition_to()``."""

    group: str | None = field(default=None, compare=False)
    """Optional group name for grouping related jobs. Included in deduplication comparison."""

    jitter: float | None = field(default=None, compare=False)
    """Upper bound, in seconds, of the random offset ``set_next_run()`` adds to ``fire_at``.

    Rolled once per occurrence, whenever ``next_run`` is set. Does not affect ``next_run``
    (unjittered logical fire time). See the ``fire_at`` field on ``Job`` for the actual
    dispatch time after jitter is applied.
    """

    timeout: float | None = field(default=None, compare=False)
    """Per-job timeout in seconds. ``None`` means use the global default
    (``config.scheduler_job_timeout_seconds``). A positive ``float`` overrides the default.
    Validated at construction: must be positive when set."""

    timeout_disabled: bool = field(default=False, compare=False)
    """When ``True``, timeout enforcement is disabled for this job regardless of the global default."""

    name: str = field(default="", compare=False)
    """Name for the job, used for identification. Required by all public Scheduler entry points
    (``Scheduler.add_job()`` and ``Scheduler.schedule()`` raise ``SchedulerNameRequiredError`` if empty)."""

    args: tuple[Any, ...] = field(default_factory=tuple, compare=False)
    """Positional arguments to pass to the job callable."""

    kwargs: dict[str, Any] = field(default_factory=dict, compare=False)
    """Keyword arguments to pass to the job callable."""

    error_handler: "SchedulerErrorHandlerType | None" = field(default=None, compare=False)
    """Optional error handler for this job.

    When set, this handler is invoked if the job raises an exception (including
    ``TimeoutError``, but excluding ``CancelledError``). Stored as-is for identity comparison
    in ``matches()``. ``compare=False`` prevents ``Callable | None`` from corrupting
    the ``@dataclass(order=True)`` heap ordering.
    """

    db_id: int | None = field(default=None, compare=False)
    """Database row ID for this job. Set by the executor after persistence; None until then."""

    source_location: str = field(default="", compare=False)
    """Captured source location (file:line) of the user code that scheduled this job."""

    registration_source: str = field(default="", compare=False)
    """Captured source code snippet of the scheduling call."""

    source_tier: SourceTier = field(default="app", compare=False)
    """Whether this job originates from a user app or the framework itself."""

    mode: ExecutionMode = field(default=ExecutionMode.SINGLE, compare=False)
    """Resolved overlap mode for this job. Determines behavior when a prior invocation is still
    running as the next occurrence becomes due. Defaults to ``ExecutionMode.SINGLE`` so existing
    direct ``Job(...)`` constructions in tests remain valid; the real resolution happens
    in ``Scheduler.schedule()``."""

    predicate: "SchedulerPredicate | None" = field(default=None, compare=False)
    """Normalized predicate callable, or ``None`` when no ``where=`` was given.

    Set once at registration time by ``Scheduler.schedule()`` — a single callable is stored
    directly, and a sequence of predicates is collapsed into a combinator. ``compare=False``
    prevents ``Callable | None`` from corrupting the ``@dataclass(order=True)`` heap ordering.
    """

    predicate_invoker: "CallableInvoker | None" = field(default=None, repr=False, compare=False)
    """Pre-built DI invoker for the predicate, or ``None`` when no ``where=`` was given.

    Built once at registration time by ``Scheduler.schedule()`` using the shared DI layer
    (``hassette.di``) and passed alongside ``predicate``. At dispatch time,
    ``predicate_invoker.invoke({Job: job})`` produces kwargs and the predicate is
    called with ``predicate(**kwargs)``."""

    guard: ExecutionModeGuard = field(init=False, compare=False)
    """Per-job overlap state machine. Created in ``__post_init__`` from ``mode``. Lives for the
    full lifetime of the job, spanning all re-fires. ``compare=False`` prevents the guard from
    affecting heap ordering. ``init=False`` — do not pass directly; set via ``mode``."""

    _scheduler: "Scheduler | None" = field(default=None, repr=False, compare=False)
    """Back-reference to the Scheduler that owns this job. Set by Scheduler.add_job()."""

    _scheduler_service: "SchedulerServiceProtocol | None" = field(default=None, repr=False, compare=False)
    """Back-reference to the SchedulerService that owns this job's runtime lifecycle.

    Set alongside ``_scheduler`` at registration time. Used by ``submit()`` to reach
    ``SchedulerService.submit_job()`` directly, without routing through the per-app
    ``Scheduler``."""

    app_error_handler_resolver: "Callable[[], SchedulerErrorHandlerType | None] | None" = field(
        default=None, init=False, repr=False, compare=False
    )
    """Closure that resolves the app-level error handler at dispatch time.

    ``compare=False`` like every other non-``sort_index`` field: functions have no ordering, so
    leaving this in the ``@dataclass(order=True)`` comparison tuple would raise ``TypeError`` on
    any ``<`` between two jobs that shared a ``sort_index``. Nothing can share one today (see
    ``__hash__``), which is exactly why it must stay out — the breakage would only appear if that
    invariant ever changed."""

    _dequeued: bool = field(default=False, repr=False, compare=False)
    """True after the job has been synchronously removed from the heap via dequeue_job()."""

    _pending_entity_time_transition: "ZonedDateTime | _WaitingSentinel | None" = field(
        default=None, repr=False, compare=False
    )
    """Event-derived EntityTime transition to apply once the job's own due fire (which popped
    it from the heap) finishes. Concrete when the entity now names a later time; ``WAITING``
    when it lost its usable time entirely. Only ever set for a ``SCHEDULED`` job caught
    mid-dispatch — see ``SchedulerService.reschedule_job()``, which applies the transition
    immediately (no pending state) for a job in any other status, since there is no
    in-flight dispatch to race in that case."""

    pending_done: "set[asyncio.Future[None]]" = field(default_factory=set, init=False, repr=False, compare=False)
    """Unresolved per-invocation completion futures for non-parallel modes.

    Each ``run_job_with_guard`` call for single/restart/queued parks the outer dispatch task on
    a future that resolves when the invocation actually runs (or is dropped/released). A queued
    invocation accepted into the guard deque has no live child until drain time, so its future
    would hang forever if the job is cancelled first. ``SchedulerService._finish_removal`` calls
    ``release_and_drain(job.guard, job.pending_done)`` so those dispatch tasks unwind.
    """

    def __hash__(self) -> int:
        # Hashing on object identity is safe: each Job is a unique object,
        # and sort_index includes id(self) as the tiebreaker, so the hash contract
        # (a == b implies hash(a) == hash(b)) holds. @dataclass(order=True) generates
        # __eq__ based on all compare=True fields (sort_index only), so two distinct
        # objects with the same sort_index are unequal and can safely share a hash bucket.
        return id(self)

    def __repr__(self) -> str:
        return f"Job(name={self.name!r}, owner_id={self.owner_id})"

    def __post_init__(self) -> None:
        if self.timeout is not None and (isinstance(self.timeout, bool) or self.timeout <= 0):
            raise ValueError("timeout must be a positive number")
        if self.timeout_disabled and self.timeout is not None:
            raise ValueError("Cannot specify both 'timeout' and 'timeout_disabled=True'")
        if self.schedule_status is ScheduleStatus.SCHEDULED and self.next_run is None:
            raise ValueError("Job(schedule_status=ScheduleStatus.SCHEDULED, ...) requires a concrete next_run")

        if self.schedule_status is not ScheduleStatus.SCHEDULED:
            # A job that has never been scheduled (WAITING, COMPLETED, MANUAL) still needs
            # some sort_index so an accidental ==/</in against it never raises AttributeError.
            # id(self) stays the tiebreaker (matching __hash__'s comment above) so two different
            # never-scheduled jobs never compare equal. Such a job must still never be inserted
            # into the heap regardless of this value — set_next_run() assigns the real ordering
            # key once the job actually transitions to SCHEDULED.
            self.sort_index = (UNSCHEDULED_SORT_KEY, id(self))

        self.guard = ExecutionModeGuard(self.mode)
        if self.next_run is not None:
            self.set_next_run(self.next_run)

        self.args = tuple(self.args)
        self.kwargs = dict(self.kwargs)

    def mark_registered(self, db_id: int) -> None:
        """Set the database ID. Called by SchedulerService.add_job() after persistence.

        First call wins — a second call keeps the original id and logs a WARNING so a
        retry or double-registration is surfaced rather than silently swallowed.
        Mirrors ``Listener.mark_registered``.
        """
        if self.db_id is not None:
            LOGGER.warning(
                "Job %s already registered with db_id=%s, ignoring new db_id=%s",
                self.name,
                self.db_id,
                db_id,
            )
            return
        self.db_id = db_id

    def matches(self, other: "Job") -> bool:
        """Check whether two jobs represent the same logical configuration.

        Compares callable, trigger (by trigger_id()), group, jitter, timeout,
        timeout_disabled, args, kwargs, error_handler (by identity), mode, and
        predicate (by equality — lambdas/closures compare by identity).
        Does not compare runtime state (db_id, next_run, sort_index, _scheduler,
        _dequeued, owner, or any other mutable runtime field).

        Two jobs with identical callable/trigger/args but different groups are distinct
        logical jobs and will not match.
        """
        if self.trigger is not None and other.trigger is not None:
            triggers_match = self.trigger.trigger_id() == other.trigger.trigger_id()
        else:
            triggers_match = self.trigger is other.trigger
        return (
            self.job == other.job
            and triggers_match
            and self.group == other.group
            and self.jitter == other.jitter
            and self.timeout == other.timeout
            and self.timeout_disabled == other.timeout_disabled
            and self.args == other.args
            and self.kwargs == other.kwargs
            and self.error_handler is other.error_handler
            and self.mode is other.mode
            and self.predicate == other.predicate
        )

    def diff_fields(self, other: "Job") -> list[str]:
        """Return a list of configuration field names that differ between two jobs.

        Compares the same fields as ``matches()`` — callable, trigger, group,
        jitter, timeout, timeout_disabled, args, kwargs, error_handler, mode,
        and predicate.
        """
        changed: list[str] = []
        if self.job != other.job:
            changed.append("job")
        self_tid = self.trigger.trigger_id() if self.trigger is not None else None
        other_tid = other.trigger.trigger_id() if other.trigger is not None else None
        if self_tid != other_tid:
            changed.append("trigger")
        if self.group != other.group:
            changed.append("group")
        if self.jitter != other.jitter:
            changed.append("jitter")
        if self.timeout != other.timeout:
            changed.append("timeout")
        if self.timeout_disabled != other.timeout_disabled:
            changed.append("timeout_disabled")
        if self.args != other.args:
            changed.append("args")
        if self.kwargs != other.kwargs:
            changed.append("kwargs")
        if self.error_handler is not other.error_handler:
            changed.append("error_handler")
        if self.mode is not other.mode:
            changed.append("mode")
        if self.predicate != other.predicate:
            changed.append("predicate")
        return changed

    def set_app_error_handler_resolver(self, resolver: "Callable[[], SchedulerErrorHandlerType | None]") -> None:
        """Set the closure that resolves the app-level error handler at dispatch time."""
        self.app_error_handler_resolver = resolver

    def remove(self) -> None:
        """Remove this job's registration by delegating to the owning Scheduler.

        Raises:
            RuntimeError: When called on a job that has not been registered with a Scheduler.
                Use ``Scheduler.remove_job(job)`` directly, or register the job first.
        """
        if self._scheduler is None:
            raise RuntimeError(
                "remove() called on a job not registered with a Scheduler. "
                "Use Scheduler.remove_job(job) or register the job first."
            )
        self._scheduler.remove_job(self)

    def submit(self) -> None:
        """Submit one manual invocation of this job through the owning SchedulerService.

        Fire-and-observe: submits through the existing overlap guard, task bucket, executor,
        and telemetry, then returns ``None`` immediately. Uses the job's registered
        args/kwargs and bypasses its predicate. Does not accept per-invocation arguments,
        and never mutates the job's automatic schedule.

        Raises:
            RuntimeError: When called on a job that has no owning SchedulerService — register
                the job via ``Scheduler.schedule()`` or ``Scheduler.register()`` first.
            JobRemovedError: When the job's registration has since been removed.
            TaskBucketSealedError: When the owning SchedulerService has been force-terminated
                and can no longer accept work.
        """
        if self._scheduler_service is None:
            raise RuntimeError(
                "submit() called on a job with no owning SchedulerService. "
                "Register the job via Scheduler.schedule() or Scheduler.register() first."
            )
        self._scheduler_service.submit_job(self)

    def set_next_run(self, next_run: ZonedDateTime | None) -> None:
        """Update the next run timestamp, fire_at, and ordering metadata.

        ``next_run`` is set to the rounded value. When the job has a non-zero ``jitter``,
        ``fire_at`` is the rounded value plus a fresh random offset in ``[0, jitter]``
        seconds; otherwise it equals ``next_run``. ``sort_index`` is built from ``fire_at``,
        so the heap orders by the actual dispatch time.

        When ``next_run`` is ``None`` (a job with no concrete automatic occurrence —
        waiting, completed, or manual), both ``next_run`` and ``fire_at`` are cleared to
        ``None`` and ``sort_index`` is left untouched — whether that leaves the
        ``UNSCHEDULED_SORT_KEY`` placeholder from ``__post_init__`` or a real key from an
        earlier scheduled run. A job in this state must never be inserted into the heap, so
        neither value is ever read for ordering.
        """
        if next_run is None:
            self.next_run = None
            self.fire_at = None
            return

        rounded = next_run.round("second")
        self.next_run = rounded
        self.fire_at = apply_jitter(rounded, self.jitter)
        self.sort_index = (self.fire_at.timestamp(unit="nanosecond"), id(self))

    def transition_to(
        self,
        status: ScheduleStatus,
        *,
        next_run: ZonedDateTime | None = None,
        fire_at: ZonedDateTime | None = None,
        reason: ScheduleStatusReason | None = None,
    ) -> None:
        """Atomically update ``schedule_status``, timing, and ``schedule_status_reason``.

        Centralizes every schedule-status transition so a call site can never forget to
        clear stale timing when leaving ``SCHEDULED``. ``SCHEDULED`` requires a concrete
        ``next_run``; every other status clears ``next_run``/``fire_at`` to ``None``
        regardless of what was passed in.

        Args:
            status: The schedule status to transition to.
            next_run: Concrete next occurrence. Required when ``status`` is ``SCHEDULED``;
                ignored (and cleared) for every other status.
            fire_at: Actual dispatch time including jitter. Defaults to the rounded
                ``next_run`` when omitted. Ignored (and cleared) for every non-``SCHEDULED``
                status.
            reason: Optional diagnostic reason. Cleared to ``None`` unless explicitly passed.

        Raises:
            ValueError: If ``status`` is ``SCHEDULED`` and ``next_run`` is ``None``.
        """
        if status is ScheduleStatus.SCHEDULED:
            if next_run is None:
                raise ValueError("transition_to(ScheduleStatus.SCHEDULED, ...) requires a concrete next_run")
            self.set_next_run(next_run)
            if fire_at is not None:
                self.fire_at = fire_at
        else:
            self.next_run = None
            self.fire_at = None

        self.schedule_status = status
        self.schedule_status_reason = reason

sort_index: tuple[int, int] = field(init=False, repr=False) class-attribute instance-attribute

Tuple of (next_run timestamp with nanoseconds, object id) for ordering in a priority queue.

Assigned the real ordering key when the job transitions to SCHEDULED with a concrete next_run (see set_next_run/transition_to). A job constructed as WAITING, COMPLETED, or MANUAL gets a placeholder key from __post_init__ instead; such a job must never be inserted into the heap, so that placeholder is never read for ordering.

owner_id: str = field(compare=False) class-attribute instance-attribute

Unique string identifier for the owner of the job, e.g., a component or integration name.

next_run: ZonedDateTime | None = field(compare=False) class-attribute instance-attribute

Unjittered logical fire time — used as previous_run in subsequent trigger calls.

None for every status other than SCHEDULED (waiting, completed, or manual-only).

fire_at: ZonedDateTime | None = field(default=None, init=False, compare=False) class-attribute instance-attribute

Actual dispatch time, including any jitter offset.

Equals next_run when no jitter is configured. Set by set_next_run(), which rolls the jitter offset at the same moment it sets next_run, so the value is final before the job is enqueued or persisted. The pop loop in _ScheduledJobQueue.pop_due_and_peek_next compares against fire_at (not next_run) to decide when to dispatch. None for every status other than SCHEDULED.

job: JobCallable = field(compare=False) class-attribute instance-attribute

The callable to execute when the job runs.

app_key: str = field(default='', compare=False) class-attribute instance-attribute

Configuration-level app key for DB registration (e.g., 'my_app'). Empty for non-App owners.

instance_index: int = field(default=0, compare=False) class-attribute instance-attribute

App instance index for DB registration. 0 for non-App owners.

instance_name: str | None = field(default=None, compare=False) class-attribute instance-attribute

App instance name, precomputed at registration time from the owning app's app_config.instance_name. None for framework-tier jobs and non-App owners.

trigger: TriggerProtocol | None = field(compare=False, default=None) class-attribute instance-attribute

The trigger that determines the job's schedule. None for a manual-only job.

schedule_status: ScheduleStatus = field(default=(ScheduleStatus.SCHEDULED), compare=False) class-attribute instance-attribute

Current schedule status. Defaults to SCHEDULED so existing direct Job(...) constructions (mostly in tests) remain valid without passing it explicitly, as long as next_run is also given — __post_init__ raises ValueError for the SCHEDULED-with-no-next_run combination, mirroring transition_to()'s guard. A WAITING/COMPLETED/MANUAL construction must pass schedule_status explicitly. Real jobs built through Scheduler.schedule()/Scheduler.register() set it explicitly, and every transition after construction routes through transition_to().

schedule_status_reason: ScheduleStatusReason | None = field(default=None, compare=False) class-attribute instance-attribute

Optional diagnostic reason qualifying schedule_status. None for a clean status with no override. Set (or cleared) only through transition_to().

group: str | None = field(default=None, compare=False) class-attribute instance-attribute

Optional group name for grouping related jobs. Included in deduplication comparison.

jitter: float | None = field(default=None, compare=False) class-attribute instance-attribute

Upper bound, in seconds, of the random offset set_next_run() adds to fire_at.

Rolled once per occurrence, whenever next_run is set. Does not affect next_run (unjittered logical fire time). See the fire_at field on Job for the actual dispatch time after jitter is applied.

timeout: float | None = field(default=None, compare=False) class-attribute instance-attribute

Per-job timeout in seconds. None means use the global default (config.scheduler_job_timeout_seconds). A positive float overrides the default. Validated at construction: must be positive when set.

timeout_disabled: bool = field(default=False, compare=False) class-attribute instance-attribute

When True, timeout enforcement is disabled for this job regardless of the global default.

name: str = field(default='', compare=False) class-attribute instance-attribute

Name for the job, used for identification. Required by all public Scheduler entry points (Scheduler.add_job() and Scheduler.schedule() raise SchedulerNameRequiredError if empty).

args: tuple[Any, ...] = field(default_factory=tuple, compare=False) class-attribute instance-attribute

Positional arguments to pass to the job callable.

kwargs: dict[str, Any] = field(default_factory=dict, compare=False) class-attribute instance-attribute

Keyword arguments to pass to the job callable.

error_handler: SchedulerErrorHandlerType | None = field(default=None, compare=False) class-attribute instance-attribute

Optional error handler for this job.

When set, this handler is invoked if the job raises an exception (including TimeoutError, but excluding CancelledError). Stored as-is for identity comparison in matches(). compare=False prevents Callable | None from corrupting the @dataclass(order=True) heap ordering.

db_id: int | None = field(default=None, compare=False) class-attribute instance-attribute

Database row ID for this job. Set by the executor after persistence; None until then.

source_location: str = field(default='', compare=False) class-attribute instance-attribute

Captured source location (file:line) of the user code that scheduled this job.

registration_source: str = field(default='', compare=False) class-attribute instance-attribute

Captured source code snippet of the scheduling call.

source_tier: SourceTier = field(default='app', compare=False) class-attribute instance-attribute

Whether this job originates from a user app or the framework itself.

mode: ExecutionMode = field(default=(ExecutionMode.SINGLE), compare=False) class-attribute instance-attribute

Resolved overlap mode for this job. Determines behavior when a prior invocation is still running as the next occurrence becomes due. Defaults to ExecutionMode.SINGLE so existing direct Job(...) constructions in tests remain valid; the real resolution happens in Scheduler.schedule().

predicate: SchedulerPredicate | None = field(default=None, compare=False) class-attribute instance-attribute

Normalized predicate callable, or None when no where= was given.

Set once at registration time by Scheduler.schedule() — a single callable is stored directly, and a sequence of predicates is collapsed into a combinator. compare=False prevents Callable | None from corrupting the @dataclass(order=True) heap ordering.

predicate_invoker: CallableInvoker | None = field(default=None, repr=False, compare=False) class-attribute instance-attribute

Pre-built DI invoker for the predicate, or None when no where= was given.

Built once at registration time by Scheduler.schedule() using the shared DI layer (hassette.di) and passed alongside predicate. At dispatch time, predicate_invoker.invoke({Job: job}) produces kwargs and the predicate is called with predicate(**kwargs).

guard: ExecutionModeGuard = field(init=False, compare=False) class-attribute instance-attribute

Per-job overlap state machine. Created in __post_init__ from mode. Lives for the full lifetime of the job, spanning all re-fires. compare=False prevents the guard from affecting heap ordering. init=False — do not pass directly; set via mode.

app_error_handler_resolver: Callable[[], SchedulerErrorHandlerType | None] | None = field(default=None, init=False, repr=False, compare=False) class-attribute instance-attribute

Closure that resolves the app-level error handler at dispatch time.

compare=False like every other non-sort_index field: functions have no ordering, so leaving this in the @dataclass(order=True) comparison tuple would raise TypeError on any < between two jobs that shared a sort_index. Nothing can share one today (see __hash__), which is exactly why it must stay out — the breakage would only appear if that invariant ever changed.

pending_done: set[asyncio.Future[None]] = field(default_factory=set, init=False, repr=False, compare=False) class-attribute instance-attribute

Unresolved per-invocation completion futures for non-parallel modes.

Each run_job_with_guard call for single/restart/queued parks the outer dispatch task on a future that resolves when the invocation actually runs (or is dropped/released). A queued invocation accepted into the guard deque has no live child until drain time, so its future would hang forever if the job is cancelled first. SchedulerService._finish_removal calls release_and_drain(job.guard, job.pending_done) so those dispatch tasks unwind.

mark_registered(db_id: int) -> None

Set the database ID. Called by SchedulerService.add_job() after persistence.

First call wins — a second call keeps the original id and logs a WARNING so a retry or double-registration is surfaced rather than silently swallowed. Mirrors Listener.mark_registered.

Source code in src/hassette/scheduler/classes.py

def mark_registered(self, db_id: int) -> None:
    """Set the database ID. Called by SchedulerService.add_job() after persistence.

    First call wins — a second call keeps the original id and logs a WARNING so a
    retry or double-registration is surfaced rather than silently swallowed.
    Mirrors ``Listener.mark_registered``.
    """
    if self.db_id is not None:
        LOGGER.warning(
            "Job %s already registered with db_id=%s, ignoring new db_id=%s",
            self.name,
            self.db_id,
            db_id,
        )
        return
    self.db_id = db_id

matches(other: Job) -> bool

Check whether two jobs represent the same logical configuration.

Compares callable, trigger (by trigger_id()), group, jitter, timeout, timeout_disabled, args, kwargs, error_handler (by identity), mode, and predicate (by equality — lambdas/closures compare by identity). Does not compare runtime state (db_id, next_run, sort_index, _scheduler, _dequeued, owner, or any other mutable runtime field).

Two jobs with identical callable/trigger/args but different groups are distinct logical jobs and will not match.

Source code in src/hassette/scheduler/classes.py

def matches(self, other: "Job") -> bool:
    """Check whether two jobs represent the same logical configuration.

    Compares callable, trigger (by trigger_id()), group, jitter, timeout,
    timeout_disabled, args, kwargs, error_handler (by identity), mode, and
    predicate (by equality — lambdas/closures compare by identity).
    Does not compare runtime state (db_id, next_run, sort_index, _scheduler,
    _dequeued, owner, or any other mutable runtime field).

    Two jobs with identical callable/trigger/args but different groups are distinct
    logical jobs and will not match.
    """
    if self.trigger is not None and other.trigger is not None:
        triggers_match = self.trigger.trigger_id() == other.trigger.trigger_id()
    else:
        triggers_match = self.trigger is other.trigger
    return (
        self.job == other.job
        and triggers_match
        and self.group == other.group
        and self.jitter == other.jitter
        and self.timeout == other.timeout
        and self.timeout_disabled == other.timeout_disabled
        and self.args == other.args
        and self.kwargs == other.kwargs
        and self.error_handler is other.error_handler
        and self.mode is other.mode
        and self.predicate == other.predicate
    )

diff_fields(other: Job) -> list[str]

Return a list of configuration field names that differ between two jobs.

Compares the same fields as matches() — callable, trigger, group, jitter, timeout, timeout_disabled, args, kwargs, error_handler, mode, and predicate.

Source code in src/hassette/scheduler/classes.py

def diff_fields(self, other: "Job") -> list[str]:
    """Return a list of configuration field names that differ between two jobs.

    Compares the same fields as ``matches()`` — callable, trigger, group,
    jitter, timeout, timeout_disabled, args, kwargs, error_handler, mode,
    and predicate.
    """
    changed: list[str] = []
    if self.job != other.job:
        changed.append("job")
    self_tid = self.trigger.trigger_id() if self.trigger is not None else None
    other_tid = other.trigger.trigger_id() if other.trigger is not None else None
    if self_tid != other_tid:
        changed.append("trigger")
    if self.group != other.group:
        changed.append("group")
    if self.jitter != other.jitter:
        changed.append("jitter")
    if self.timeout != other.timeout:
        changed.append("timeout")
    if self.timeout_disabled != other.timeout_disabled:
        changed.append("timeout_disabled")
    if self.args != other.args:
        changed.append("args")
    if self.kwargs != other.kwargs:
        changed.append("kwargs")
    if self.error_handler is not other.error_handler:
        changed.append("error_handler")
    if self.mode is not other.mode:
        changed.append("mode")
    if self.predicate != other.predicate:
        changed.append("predicate")
    return changed

set_app_error_handler_resolver(resolver: Callable[[], SchedulerErrorHandlerType | None]) -> None

Set the closure that resolves the app-level error handler at dispatch time.

Source code in src/hassette/scheduler/classes.py

def set_app_error_handler_resolver(self, resolver: "Callable[[], SchedulerErrorHandlerType | None]") -> None:
    """Set the closure that resolves the app-level error handler at dispatch time."""
    self.app_error_handler_resolver = resolver

remove() -> None

Remove this job's registration by delegating to the owning Scheduler.

Raises:

Type	Description
RuntimeError	When called on a job that has not been registered with a Scheduler. Use Scheduler.remove_job(job) directly, or register the job first.

Source code in src/hassette/scheduler/classes.py

def remove(self) -> None:
    """Remove this job's registration by delegating to the owning Scheduler.

    Raises:
        RuntimeError: When called on a job that has not been registered with a Scheduler.
            Use ``Scheduler.remove_job(job)`` directly, or register the job first.
    """
    if self._scheduler is None:
        raise RuntimeError(
            "remove() called on a job not registered with a Scheduler. "
            "Use Scheduler.remove_job(job) or register the job first."
        )
    self._scheduler.remove_job(self)

submit() -> None

Submit one manual invocation of this job through the owning SchedulerService.

Fire-and-observe: submits through the existing overlap guard, task bucket, executor, and telemetry, then returns None immediately. Uses the job's registered args/kwargs and bypasses its predicate. Does not accept per-invocation arguments, and never mutates the job's automatic schedule.

Raises:

Type	Description
RuntimeError	When called on a job that has no owning SchedulerService — register the job via Scheduler.schedule() or Scheduler.register() first.
JobRemovedError	When the job's registration has since been removed.
TaskBucketSealedError	When the owning SchedulerService has been force-terminated and can no longer accept work.

Source code in src/hassette/scheduler/classes.py

def submit(self) -> None:
    """Submit one manual invocation of this job through the owning SchedulerService.

    Fire-and-observe: submits through the existing overlap guard, task bucket, executor,
    and telemetry, then returns ``None`` immediately. Uses the job's registered
    args/kwargs and bypasses its predicate. Does not accept per-invocation arguments,
    and never mutates the job's automatic schedule.

    Raises:
        RuntimeError: When called on a job that has no owning SchedulerService — register
            the job via ``Scheduler.schedule()`` or ``Scheduler.register()`` first.
        JobRemovedError: When the job's registration has since been removed.
        TaskBucketSealedError: When the owning SchedulerService has been force-terminated
            and can no longer accept work.
    """
    if self._scheduler_service is None:
        raise RuntimeError(
            "submit() called on a job with no owning SchedulerService. "
            "Register the job via Scheduler.schedule() or Scheduler.register() first."
        )
    self._scheduler_service.submit_job(self)

set_next_run(next_run: ZonedDateTime | None) -> None

Update the next run timestamp, fire_at, and ordering metadata.

next_run is set to the rounded value. When the job has a non-zero jitter, fire_at is the rounded value plus a fresh random offset in [0, jitter] seconds; otherwise it equals next_run. sort_index is built from fire_at, so the heap orders by the actual dispatch time.

When next_run is None (a job with no concrete automatic occurrence — waiting, completed, or manual), both next_run and fire_at are cleared to None and sort_index is left untouched — whether that leaves the UNSCHEDULED_SORT_KEY placeholder from __post_init__ or a real key from an earlier scheduled run. A job in this state must never be inserted into the heap, so neither value is ever read for ordering.

Source code in src/hassette/scheduler/classes.py

def set_next_run(self, next_run: ZonedDateTime | None) -> None:
    """Update the next run timestamp, fire_at, and ordering metadata.

    ``next_run`` is set to the rounded value. When the job has a non-zero ``jitter``,
    ``fire_at`` is the rounded value plus a fresh random offset in ``[0, jitter]``
    seconds; otherwise it equals ``next_run``. ``sort_index`` is built from ``fire_at``,
    so the heap orders by the actual dispatch time.

    When ``next_run`` is ``None`` (a job with no concrete automatic occurrence —
    waiting, completed, or manual), both ``next_run`` and ``fire_at`` are cleared to
    ``None`` and ``sort_index`` is left untouched — whether that leaves the
    ``UNSCHEDULED_SORT_KEY`` placeholder from ``__post_init__`` or a real key from an
    earlier scheduled run. A job in this state must never be inserted into the heap, so
    neither value is ever read for ordering.
    """
    if next_run is None:
        self.next_run = None
        self.fire_at = None
        return

    rounded = next_run.round("second")
    self.next_run = rounded
    self.fire_at = apply_jitter(rounded, self.jitter)
    self.sort_index = (self.fire_at.timestamp(unit="nanosecond"), id(self))

transition_to(status: ScheduleStatus, *, next_run: ZonedDateTime | None = None, fire_at: ZonedDateTime | None = None, reason: ScheduleStatusReason | None = None) -> None

Atomically update schedule_status, timing, and schedule_status_reason.

Centralizes every schedule-status transition so a call site can never forget to clear stale timing when leaving SCHEDULED. SCHEDULED requires a concrete next_run; every other status clears next_run/fire_at to None regardless of what was passed in.

Parameters:

Name	Type	Description	Default
status	ScheduleStatus	The schedule status to transition to.	required
next_run	ZonedDateTime | None	Concrete next occurrence. Required when status is SCHEDULED; ignored (and cleared) for every other status.	None
fire_at	ZonedDateTime | None	Actual dispatch time including jitter. Defaults to the rounded next_run when omitted. Ignored (and cleared) for every non-SCHEDULED status.	None
reason	ScheduleStatusReason | None	Optional diagnostic reason. Cleared to None unless explicitly passed.	None

Raises:

Type	Description
ValueError	If status is SCHEDULED and next_run is None.

Source code in src/hassette/scheduler/classes.py

def transition_to(
    self,
    status: ScheduleStatus,
    *,
    next_run: ZonedDateTime | None = None,
    fire_at: ZonedDateTime | None = None,
    reason: ScheduleStatusReason | None = None,
) -> None:
    """Atomically update ``schedule_status``, timing, and ``schedule_status_reason``.

    Centralizes every schedule-status transition so a call site can never forget to
    clear stale timing when leaving ``SCHEDULED``. ``SCHEDULED`` requires a concrete
    ``next_run``; every other status clears ``next_run``/``fire_at`` to ``None``
    regardless of what was passed in.

    Args:
        status: The schedule status to transition to.
        next_run: Concrete next occurrence. Required when ``status`` is ``SCHEDULED``;
            ignored (and cleared) for every other status.
        fire_at: Actual dispatch time including jitter. Defaults to the rounded
            ``next_run`` when omitted. Ignored (and cleared) for every non-``SCHEDULED``
            status.
        reason: Optional diagnostic reason. Cleared to ``None`` unless explicitly passed.

    Raises:
        ValueError: If ``status`` is ``SCHEDULED`` and ``next_run`` is ``None``.
    """
    if status is ScheduleStatus.SCHEDULED:
        if next_run is None:
            raise ValueError("transition_to(ScheduleStatus.SCHEDULED, ...) requires a concrete next_run")
        self.set_next_run(next_run)
        if fire_at is not None:
            self.fire_at = fire_at
    else:
        self.next_run = None
        self.fire_at = None

    self.schedule_status = status
    self.schedule_status_reason = reason

apply_jitter(fire_time: ZonedDateTime, jitter: float | None) -> ZonedDateTime

Return fire_time plus a random offset in [0, jitter] seconds.

Returns fire_time unchanged when jitter is None or zero, or when the offset would push it out of the representable range.

Source code in src/hassette/scheduler/classes.py

def apply_jitter(fire_time: ZonedDateTime, jitter: float | None) -> ZonedDateTime:
    """Return ``fire_time`` plus a random offset in ``[0, jitter]`` seconds.

    Returns ``fire_time`` unchanged when ``jitter`` is ``None`` or zero, or when the offset
    would push it out of the representable range.
    """
    if not jitter:
        return fire_time
    try:
        return fire_time.add(seconds=random.uniform(0, jitter))
    except ValueError:
        return fire_time