Command Executor

CommandExecutor service: executes handler invocations and scheduled jobs with telemetry.

ExecutionMarker dataclass

Immutable snapshot of the execution currently holding the loop thread.

Published to CommandExecutor.current_execution via a single atomic attribute assignment so an off-loop watchdog thread can read it without a lock. A separate OS thread cannot read another thread's ContextVar, so the blocking-IO watchdog reads this plain attribute instead of CURRENT_EXECUTION_ID. Frozen (immutable) so a reader can never observe a half-mutated marker — replacement is always a whole-object rebind.

Single-slot semantics: the marker reflects the most recent bind_execution_context that has not yet been unbound. While the loop thread is frozen by a blocking call no interleaving occurs, so the marker usually names the execution that froze it. But a handler that yields (await) and is then displaced by another execution leaves a stale marker: the loop is responsive across the await, yet a later synchronous block by the displacing execution (or by framework code) makes the single slot name the wrong owner. task_id guards against this — a reader compares it against the task actually running on the loop during the freeze and refuses to attribute when they differ (see LoopWatchdog and the Tier 2 guard).

started_at is time.monotonic() (process-wide, readable across threads) for measuring how long the execution has held the loop. task_id is id(asyncio.current_task()) captured at bind — the identity of the asyncio Task that owns this execution.

Source code in src/hassette/core/command_executor.py

@dataclass(frozen=True)
class ExecutionMarker:
    """Immutable snapshot of the execution currently holding the loop thread.

    Published to ``CommandExecutor.current_execution`` via a single atomic attribute
    assignment so an off-loop watchdog thread can read it without a lock. A separate OS
    thread cannot read another thread's ``ContextVar``, so the blocking-IO watchdog reads
    this plain attribute instead of ``CURRENT_EXECUTION_ID``. Frozen (immutable) so a reader
    can never observe a half-mutated marker — replacement is always a whole-object rebind.

    Single-slot semantics: the marker reflects the most recent ``bind_execution_context``
    that has not yet been unbound. While the loop thread is *frozen* by a blocking call no
    interleaving occurs, so the marker usually names the execution that froze it. But a handler
    that yields (``await``) and is then displaced by another execution leaves a stale marker:
    the loop is responsive across the ``await``, yet a later synchronous block by the *displacing*
    execution (or by framework code) makes the single slot name the wrong owner. ``task_id`` guards
    against this — a reader compares it against the task actually running on the loop during the
    freeze and refuses to attribute when they differ (see ``LoopWatchdog`` and the Tier 2 guard).

    ``started_at`` is ``time.monotonic()`` (process-wide, readable across threads) for measuring
    how long the execution has held the loop. ``task_id`` is ``id(asyncio.current_task())`` captured
    at bind — the identity of the asyncio Task that owns this execution.
    """

    app_key: str | None
    instance_name: str | None
    execution_id: str
    started_at: float
    instance_index: int | None = None
    task_id: int | None = None
    """``id()`` of the owning ``asyncio.Task``, or ``None`` if bound outside a task."""

task_id: int | None = None class-attribute instance-attribute

id() of the owning asyncio.Task, or None if bound outside a task.

RetryableBatch dataclass

A batch of records that failed to persist and should be retried.

Attributes:

Name	Type	Description
records	list[ExecutionRecord]	Unified execution records to retry.
retry_count	int	Number of times this whole batch has been retried by the executor. Unrelated to ExecutionRecord.retry_count (a per-row schema column that is currently always 0); this one drives the in-memory retry/backoff loop.
not_before	float	Monotonic timestamp (time.monotonic()) before which this batch must not be retried. Zero means eligible immediately.

Source code in src/hassette/core/command_executor.py

@dataclass
class RetryableBatch:
    """A batch of records that failed to persist and should be retried.

    Attributes:
        records: Unified execution records to retry.
        retry_count: Number of times this whole batch has been retried by the executor.
            Unrelated to ``ExecutionRecord.retry_count`` (a per-row schema column that is
            currently always 0); this one drives the in-memory retry/backoff loop.
        not_before: Monotonic timestamp (time.monotonic()) before which this batch
            must not be retried. Zero means eligible immediately.
    """

    records: list[ExecutionRecord] = field(default_factory=list)
    retry_count: int = 0
    not_before: float = 0.0

CommandExecutor

Bases: Service

Executes handler invocations and scheduled jobs, persisting execution records to SQLite.

Lifecycle

depends_on: DatabaseService (auto-waited before lifecycle hooks). serve(): drains the write queue in batches until shutdown, then flushes remaining records.

Records are queued immediately after each execution and persisted in batches by serve(). On shutdown, flush_queue() persists any remaining records before returning.

Source code in src/hassette/core/command_executor.py

class CommandExecutor(Service):
    """Executes handler invocations and scheduled jobs, persisting execution records to SQLite.

    Lifecycle:
        depends_on: DatabaseService (auto-waited before lifecycle hooks).
        serve(): drains the write queue in batches until shutdown, then flushes remaining records.

    Records are queued immediately after each execution and persisted in batches by serve().
    On shutdown, flush_queue() persists any remaining records before returning.
    """

    depends_on: ClassVar[list[type[Resource]]] = [DatabaseService]
    restart_spec: ClassVar[RestartSpec] = RestartSpec(
        restart_type=RestartType.TRANSIENT,
        budget_intensity=3,
        budget_period_seconds=120,
    )

    _write_queue: asyncio.Queue[ExecutionRecord | RetryableBatch]
    """Bounded queue of execution records pending DB persistence."""

    repository: TelemetryRepository
    """Repository for all telemetry SQL writes."""

    _dropped_overflow: int
    """Count of records dropped because the write queue was full."""

    _dropped_exhausted: int
    """Count of records dropped because retry_count exceeded the maximum."""

    _dropped_shutdown: int
    """Count of records dropped during shutdown flush (DB unavailable)."""

    _error_handler_failures: int
    """Count of user-registered error handler invocations that raised an exception or timed out."""

    _last_capacity_warn_ts: float
    """Monotonic timestamp of the last 75%-capacity warning (rate-limiting)."""

    _timeout_warn_timestamps: dict[int, float]
    """Per-entity timeout warning rate limiter.

    Maps in-memory ID (listener_id or job_id) to the monotonic timestamp of the
    last timeout WARNING. Entries older than 60s are lazily evicted during
    rate-limit checks.
    """

    current_execution: ExecutionMarker | None = None
    """Thread-visible marker of the execution currently on the loop thread, or None when idle.

    Read by the off-loop blocking-IO watchdog (Tier 1) to attribute a loop freeze to the app
    that caused it. Published via atomic attribute assignment in ``bind_execution_context`` and
    cleared in ``unbind_execution_context``. Each assignment creates an instance-level attribute
    that shadows this class-level default, so instances never share marker state. The class-level
    ``None`` default exists only so the attribute is safe to read before the first execution and
    on instances built via ``__new__`` in test helpers (which bypass ``__init__`` — see
    ``tests/unit/core/conftest.py``).
    """

    def __init__(self, hassette: "Hassette", *, parent: "Resource | None" = None) -> None:
        super().__init__(hassette, parent=parent)
        self._write_queue = asyncio.Queue(maxsize=hassette.config.database.telemetry_write_queue_max)
        self.repository = TelemetryRepository(hassette.database_service)
        self._dropped_overflow = 0
        self._dropped_exhausted = 0
        self._dropped_shutdown = 0
        self._error_handler_failures = 0
        self._last_capacity_warn_ts = 0.0
        self._timeout_warn_timestamps = {}

    @property
    def config_log_level(self) -> LOG_LEVEL_TYPE:
        return self.hassette.config.logging.command_executor

    async def serve(self) -> None:
        """Drain the write queue in batches until shutdown, then flush remaining records.

        Uses asyncio.wait() with a timeout equal to max_flush_interval_seconds so that records
        never sit in the queue longer than that interval, even if the batch size
        threshold is not reached.  When the timeout fires (done is empty), whatever
        is currently in the queue is drained immediately.
        """
        mark_ready(self, reason="CommandExecutor started")
        flush_interval = self.hassette.config.database.max_flush_interval_seconds

        while True:
            get_fut = asyncio.create_task(self._write_queue.get())
            shutdown_fut = asyncio.create_task(self.shutdown_event.wait())

            done, pending = await asyncio.wait(
                [get_fut, shutdown_fut],
                timeout=flush_interval,
                return_when=asyncio.FIRST_COMPLETED,
            )

            # Cancel the pending futures to avoid task leaks
            for fut in pending:
                fut.cancel()
                with contextlib.suppress(asyncio.CancelledError, Exception):
                    await fut

            if self.shutdown_event.is_set():
                # get_fut dequeued an item before shutdown was detected — re-enqueue so flush_queue() picks it up
                if get_fut in done and not get_fut.cancelled() and get_fut.exception() is None:
                    result = get_fut.result()
                    try:
                        self._write_queue.put_nowait(result)
                    except asyncio.QueueFull:
                        self._dropped_overflow += 1
                        self.logger.error(
                            "Write queue full during shutdown — dropping 1 record (total dropped: %d)",
                            self._dropped_overflow,
                        )
                await self.flush_queue()
                return

            if get_fut in done and not get_fut.cancelled() and get_fut.exception() is None:
                # An item arrived — drain the full queue in one batch
                try:
                    await self.drain_and_persist(first_item=get_fut.result())
                except Exception:
                    self.logger.exception(
                        "drain_and_persist failed — records from this batch are dropped (already dequeued)"
                    )
            elif not done:
                # Timeout — timer fired; drain whatever accumulated without a triggering item
                if not self._write_queue.empty():
                    try:
                        await self.drain_and_persist()
                    except Exception:
                        self.logger.exception(
                            "drain_and_persist failed (timer flush) — records from this batch may be dropped"
                        )

    def get_drop_counters(self) -> tuple[int, int, int]:
        """Return (dropped_overflow, dropped_exhausted, dropped_shutdown) counters.

        Returns:
            A tuple of counters where:
            - overflow_count: records dropped because the write queue was full.
            - exhausted_count: records dropped because max retries were exceeded.
            - shutdown_count: records dropped during shutdown flush.
        """
        return (self._dropped_overflow, self._dropped_exhausted, self._dropped_shutdown)

    def get_error_handler_failures(self) -> int:
        """Return the count of user error handler invocations that raised or timed out.

        Incremented each time a user-registered error handler (bus or scheduler)
        raises an exception or times out during invocation.

        Returns:
            The cumulative error handler failure count for this session.
        """
        return self._error_handler_failures

    async def execute(self, cmd: InvokeHandler | ExecuteJob) -> None:
        """Execute a command (handler invocation or scheduled job).

        Args:
            cmd: The command to execute.
        """
        match cmd:
            case InvokeHandler():
                await self.execute_handler(cmd)
            case ExecuteJob():
                await self.execute_job(cmd)

    async def _execute(
        self,
        fn: Callable[[], Awaitable[None]],
        cmd: InvokeHandler | ExecuteJob,
        log_error: Callable[[ExecutionResult], None],
        execution_id: str,
    ) -> ExecutionResult:
        """Core execution wrapper: time the call, capture errors, queue the record.

        Wraps ``track_execution()`` with a tier-aware exception contract:

        - ``CancelledError``   — record queued with status='cancelled', then re-raised.
        - ``TimeoutError``     — record queued with status='timed_out', then swallowed.
                                 Warning logged by ``log_timeout_rate_limited``; not re-raised.
        - ``DependencyError``  — app tier: no traceback; framework tier: traceback included.
        - ``HassetteError``    — app tier: no traceback; framework tier: traceback included.
        - ``Exception``        — record queued with status='error', traceback included.
        - success              — record queued with status='success'.

        ``result`` is initialized before the ``async with`` block so that a
        ``CancelledError`` raised before ``track_execution()`` yields still has a
        safe default to queue.

        Args:
            fn: The async callable to execute (a zero-argument coroutine factory).
            cmd: The originating command, used to build the record in callers.
            log_error: A callback that logs the error details from the result.
                Called for error paths (not cancelled, not success).
            execution_id: The UUIDv7 string generated by the calling method
                (execute_handler or execute_job) for this execution.

        Returns:
            The populated ``ExecutionResult``.
        """
        execution_start_ts = time.time()
        result = ExecutionResult(execution_id=execution_id, status="cancelled")
        match cmd.source_tier:
            case "app":
                known: tuple[type[Exception], ...] = (DependencyError, HassetteError)
            case "framework":
                known = ()
            case _:
                raise AssertionError(f"Unexpected source_tier: {cmd.source_tier!r}")
        try:
            async with track_execution(known_errors=known) as result:
                result.execution_id = execution_id
                async with asyncio.timeout(cmd.effective_timeout):
                    await fn()
        except asyncio.CancelledError:
            self.enqueue_record(self.build_record(cmd, result, execution_start_ts, execution_id))
            raise
        except Exception:  # noqa: S110 — intentional: ExecutionResult is populated and error logged upstream
            pass
        # result is available for both success and error paths
        if result.is_timed_out:
            # Check whether the sync worker thread is still running the submitted fn.
            # The handle was set by run_in_thread on this same asyncio task (same context),
            # so SYNC_WORKER_HANDLE.get() returns the same SyncWorkerHandle the worker mutates.
            # Three non-leak cases: (a) handle.thread is None — async handler or not-started
            # timeout; (b) handle.active is False — fn finished just before the timeout check,
            # but the pool thread is still alive (pool threads persist between jobs); (c) the
            # thread is no longer alive. Only an active, live thread is a genuine leak.
            handle = SYNC_WORKER_HANDLE.get()
            if handle is not None and handle.active and handle.thread is not None and handle.thread.is_alive():
                result.thread_leaked = True
                self.logger.warning(
                    "Sync worker thread still alive after timeout (%.1fms elapsed, thread=%s) — "
                    "worker will run to completion on the dedicated executor",
                    result.duration_ms,
                    handle.thread.name,
                )
            if cmd.effective_timeout is not None:
                self.log_timeout_rate_limited(cmd, result)
            else:
                self.logger.warning(
                    "Handler raised TimeoutError after %.1fms (no framework timeout configured — "
                    "exception originated from user code)",
                    result.duration_ms,
                )
        # Clear the handle unconditionally so a future invocation that reuses this asyncio
        # context cannot read a stale worker reference and report a false thread_leaked.
        SYNC_WORKER_HANDLE.set(None)
        if result.is_error:
            log_error(result)
        self.enqueue_record(self.build_record(cmd, result, execution_start_ts, execution_id))
        return result

    def log_timeout_rate_limited(self, cmd: InvokeHandler | ExecuteJob, result: ExecutionResult) -> None:
        """Log a timeout WARNING, rate-limited per entity (60s suppression window).

        Uses the in-memory ID (``listener_id`` for handlers, object identity for jobs)
        to key the suppression window. Lazily evicts stale entries (>60s old)
        during each check.
        """
        now = time.monotonic()

        # Determine the in-memory ID for rate-limiting
        match cmd:
            case InvokeHandler():
                entity_id = cmd.listener.listener_id
                label = f"listener_id={cmd.listener.listener_id}, topic={cmd.topic}"
            case ExecuteJob():
                entity_id = id(cmd.job)
                label = f"job_db_id={cmd.job_db_id}, name={cmd.job.name}"

        # Lazy eviction of stale entries, then cap to bound memory under sustained unavailability
        stale_ids = [k for k, ts in self._timeout_warn_timestamps.items() if now - ts > _TIMEOUT_WARN_SUPPRESS_SECS]
        for k in stale_ids:
            del self._timeout_warn_timestamps[k]
        if len(self._timeout_warn_timestamps) > _TIMEOUT_WARN_CACHE_MAX:
            self._timeout_warn_timestamps.clear()

        # Rate-limit check
        last_ts = self._timeout_warn_timestamps.get(entity_id)
        if last_ts is not None and now - last_ts < _TIMEOUT_WARN_SUPPRESS_SECS:
            return  # suppressed
        self._timeout_warn_timestamps[entity_id] = now

        self.logger.warning(
            "Execution timed out after %.1fms (%s, timeout=%.1fs)",
            result.duration_ms,
            label,
            cmd.effective_timeout,
        )

    def enqueue_record(self, record: ExecutionRecord) -> None:
        """Enqueue a record, dropping and logging if the queue is full.

        Also logs a WARNING when the queue exceeds 75% capacity (rate-limited).
        """
        max_size = self._write_queue.maxsize
        current_size = self._write_queue.qsize()

        # 75% capacity warning (rate-limited)
        if max_size > 0 and current_size >= int(max_size * _CAPACITY_WARN_THRESHOLD):
            now = time.monotonic()
            if now - self._last_capacity_warn_ts >= _CAPACITY_WARN_RATE_LIMIT_SECS:
                self._last_capacity_warn_ts = now
                self.logger.warning(
                    "Write queue at %d/%d (%.0f%%) — high telemetry load",
                    current_size,
                    max_size,
                    (current_size / max_size) * 100,
                )

        try:
            self._write_queue.put_nowait(record)
        except asyncio.QueueFull:
            self._dropped_overflow += 1
            self.logger.error(
                "Write queue full (%d/%d) — dropping record (total dropped: %d)",
                current_size,
                max_size,
                self._dropped_overflow,
            )

    def build_record(
        self,
        cmd: InvokeHandler | ExecuteJob,
        result: ExecutionResult,
        execution_start_ts: float,
        execution_id: str,
    ) -> ExecutionRecord:
        """Build a unified ExecutionRecord from the execution result and command.

        session_id is set to None if the session hasn't been created yet (pre-Phase 1).
        The actual session_id is injected at drain time in persist_batch.

        Args:
            cmd: The originating command.
            result: The execution result with timing and error info.
            execution_start_ts: Unix timestamp when execution began.
            execution_id: UUIDv7 string for this execution instance.
        """
        session_id = self.hassette.try_session_id()

        match cmd:
            case InvokeHandler():
                return ExecutionRecord(
                    kind="handler",
                    listener_id=cmd.listener_id,
                    job_id=None,
                    session_id=session_id,
                    execution_start_ts=execution_start_ts,
                    duration_ms=result.duration_ms,
                    status=result.status,
                    app_key=cmd.listener.identity.app_key,
                    instance_index=cmd.listener.identity.instance_index,
                    source_tier=cmd.source_tier,
                    is_di_failure=result.is_di_failure,
                    thread_leaked=result.thread_leaked,
                    error_type=result.error_type,
                    error_message=result.error_message,
                    error_traceback=result.error_traceback,
                    execution_id=execution_id,
                    trigger_context_id=None if cmd.is_synthetic else cmd.event.payload.event_id,
                    trigger_origin=SYNTHETIC_ORIGIN if cmd.is_synthetic else cmd.event.payload.origin,
                )
            case ExecuteJob():
                return ExecutionRecord(
                    kind="job",
                    listener_id=None,
                    job_id=cmd.job_db_id,
                    session_id=session_id,
                    execution_start_ts=execution_start_ts,
                    duration_ms=result.duration_ms,
                    status=result.status,
                    app_key=cmd.job.app_key,
                    instance_index=cmd.job.instance_index,
                    source_tier=cmd.source_tier,
                    is_di_failure=result.is_di_failure,
                    thread_leaked=result.thread_leaked,
                    error_type=result.error_type,
                    error_message=result.error_message,
                    error_traceback=result.error_traceback,
                    execution_id=execution_id,
                    trigger_mode=cmd.trigger_mode,
                )

    def bind_execution_context(
        self,
        app_key: str | None,
        instance_index: int,
        instance_name: str | None,
        *,
        execution_kind: str | None = None,
        listener_id: int | None = None,
        job_id: int | None = None,
    ) -> tuple[str, Token[str | None]]:
        """Set CURRENT_EXECUTION_ID and bind structlog context vars for the duration of an execution."""
        execution_id = str(uuid_utils.uuid7())
        token = CURRENT_EXECUTION_ID.set(execution_id)
        resolved_app_key = app_key or None
        structlog.contextvars.bind_contextvars(
            app_key=resolved_app_key,
            instance_name=instance_name,
            instance_index=instance_index,
            execution_kind=execution_kind,
            listener_id=listener_id,
            job_id=job_id,
        )
        # Capture the owning task identity so a cross-thread reader can confirm this marker
        # names the task actually frozen on the loop, not a displaced one. This runs inside an
        # execute_handler/execute_job task in production; guard the no-running-loop case so a
        # direct synchronous call (tests) does not raise.
        try:
            current_task = asyncio.current_task()
        except RuntimeError:
            current_task = None
        # Publish the thread-visible marker last, as a single atomic assignment, so the off-loop
        # watchdog reads a fully-formed snapshot of the execution now holding the loop thread.
        self.current_execution = ExecutionMarker(
            app_key=resolved_app_key,
            instance_name=instance_name,
            execution_id=execution_id,
            started_at=time.monotonic(),
            instance_index=instance_index,
            task_id=id(current_task) if current_task is not None else None,
        )
        return execution_id, token

    def unbind_execution_context(self, token: Token[str | None]) -> None:
        """Clear the execution marker and reset the ContextVar/structlog binding on exit.

        Paired with ``bind_execution_context``. Was a ``@staticmethod``; it takes ``self`` now
        so it can clear ``current_execution``. Clearing the marker first (to ``None`` = idle) is
        deliberate: the off-loop watchdog reads only ``current_execution``, so a clear-first order
        means it can never attribute a stale execution once this returns. ``CURRENT_EXECUTION_ID``
        is thread-local, so its reset order relative to the marker is invisible to other threads.
        """
        self.current_execution = None
        CURRENT_EXECUTION_ID.reset(token)
        structlog.contextvars.unbind_contextvars(
            "app_key", "instance_name", "instance_index", "execution_kind", "listener_id", "job_id"
        )

    async def execute_handler(self, cmd: InvokeHandler) -> None:
        """Execute a listener handler invocation and queue the result record."""
        execution_id, token = self.bind_execution_context(
            cmd.listener.identity.app_key,
            cmd.listener.identity.instance_index,
            cmd.listener.identity.instance_name,
            execution_kind="handler",
            listener_id=cmd.listener_id,
        )
        try:

            def log_error(result: ExecutionResult) -> None:
                if result.error_traceback is None:
                    self.logger.error(
                        "Handler error (topic=%s, exec=%s): %s", cmd.topic, execution_id, result.error_message
                    )
                else:
                    self.logger.error(
                        "Handler error (topic=%s, handler=%r, exec=%s)\n%s",
                        cmd.topic,
                        cmd.listener,
                        execution_id,
                        result.error_traceback,
                    )

            result = await self._execute(lambda: cmd.listener.invoker.invoke(cmd.event), cmd, log_error, execution_id)

            if (result.is_error or result.is_timed_out) and result.exc is not None:
                error_handler = cmd.listener.invoker.error_handler or cmd.app_level_error_handler
                if error_handler is not None:
                    ctx = BusErrorContext(
                        exception=result.exc,
                        traceback="".join(traceback.format_exception(result.exc)),
                        execution_id=execution_id,
                        topic=cmd.topic,
                        listener_name=repr(cmd.listener),
                        event=cmd.event,
                    )
                    # FIXME(#573): no per-listener rate-limit on error handler spawns — high-frequency
                    # failures can accumulate unbounded concurrent tasks.
                    self.task_bucket.spawn(
                        self.invoke_error_handler(error_handler, ctx),
                        name="executor:bus_error_handler",
                    )
        finally:
            self.unbind_execution_context(token)

    async def execute_job(self, cmd: ExecuteJob) -> None:
        """Execute a scheduled job and queue the result record."""
        execution_id, token = self.bind_execution_context(
            cmd.job.app_key,
            cmd.job.instance_index,
            cmd.job.instance_name,
            execution_kind="job",
            job_id=cmd.job_db_id,
        )
        try:

            def log_error(result: ExecutionResult) -> None:
                if result.error_traceback is None:
                    self.logger.error(
                        "Job error (job_db_id=%s, exec=%s): %s", cmd.job_db_id, execution_id, result.error_message
                    )
                else:
                    self.logger.error(
                        "Job error (job_db_id=%s, exec=%s)\n%s", cmd.job_db_id, execution_id, result.error_traceback
                    )

            result = await self._execute(cmd.callable, cmd, log_error, execution_id)

            if (result.is_error or result.is_timed_out) and result.exc is not None:
                error_handler = cmd.job.error_handler or cmd.app_level_error_handler
                if error_handler is not None:
                    ctx = SchedulerErrorContext(
                        exception=result.exc,
                        traceback="".join(traceback.format_exception(result.exc)),
                        execution_id=execution_id,
                        job_name=cmd.job.name,
                        job_group=cmd.job.group,
                        args=cmd.job.args,
                        kwargs=dict(cmd.job.kwargs),
                    )
                    # FIXME(#573): no per-job rate-limit on error handler spawns.
                    self.task_bucket.spawn(
                        self.invoke_error_handler(error_handler, ctx),
                        name="executor:scheduler_error_handler",
                    )
        finally:
            self.unbind_execution_context(token)

    async def invoke_error_handler(
        self,
        handler: "Callable",
        ctx: ErrorContext,
    ) -> None:
        """Invoke a user-registered error handler in a separate spawned task.

        Note: CURRENT_EXECUTION_ID is set to the parent execution's ID via inherited
        context snapshot. This is intentional for error correlation but is not reset
        here — sub-tasks spawned by user error handler code will inherit the same ID.
        """
        async_handler = self.task_bucket.make_async_adapter(handler)
        timeout = self.hassette.config.lifecycle.error_handler_timeout_seconds
        label = ctx.log_label
        try:
            async with asyncio.timeout(timeout):
                await async_handler(ctx)
        except TimeoutError:
            self._error_handler_failures += 1
            if timeout is None:
                self.logger.exception("Error handler raised TimeoutError (%s)", label)
            else:
                self.logger.warning("Error handler timed out after %.1fs (%s)", timeout, label)
        except Exception:
            self._error_handler_failures += 1
            self.logger.exception("Error handler raised an exception (%s)", label)

    async def register_listener(self, registration: ListenerRegistration) -> int:
        """Insert a listener registration into the listeners table.

        Args:
            registration: The listener registration data.

        Returns:
            The row ID of the inserted row.
        """
        listener_id = await self.hassette.database_service.submit(self.repository.register_listener(registration))
        return listener_id

    async def register_job(self, registration: ScheduledJobRegistration) -> int:
        """Insert a scheduled job registration into the scheduled_jobs table.

        Args:
            registration: The scheduled job registration data.

        Returns:
            The row ID of the inserted row.
        """
        job_id = await self.hassette.database_service.submit(self.repository.register_job(registration))
        return job_id

    async def mark_job_cancelled(self, db_id: int) -> None:
        """Set ``cancelled_at`` on the scheduled_jobs row to persist durable cancellation state.

        Delegates to ``TelemetryRepository.mark_job_cancelled`` via ``DatabaseService.submit``.

        Args:
            db_id: The ``id`` of the ``scheduled_jobs`` row to mark as cancelled.
        """
        await self.hassette.database_service.submit(self.repository.mark_job_cancelled(db_id))

    async def mark_listener_cancelled(self, db_id: int) -> None:
        """Set ``cancelled_at`` on the listeners row to persist durable cancellation state.

        Delegates to ``TelemetryRepository.mark_listener_cancelled`` via ``DatabaseService.submit``.

        Args:
            db_id: The ``id`` of the ``listeners`` row to mark as cancelled.
        """
        await self.hassette.database_service.submit(self.repository.mark_listener_cancelled(db_id))

    async def reconcile_registrations(
        self,
        app_key: str,
        live_listener_ids: list[int],
        live_job_ids: list[int],
        *,
        session_id: int | None = None,
    ) -> None:
        """Reconcile listener and job registrations for an app after initialization.

        Deletes stale rows without history, sets ``retired_at`` on stale rows with
        history, and deletes ``once=True`` rows from previous sessions. Delegates
        to ``TelemetryRepository.reconcile_registrations`` via ``DatabaseService.submit``.

        Args:
            app_key: The app key to reconcile.
            live_listener_ids: IDs of currently active listener rows.
            live_job_ids: IDs of currently active scheduled_job rows.
            session_id: Current session ID, used to guard once=True row deletion.
        """
        await self.hassette.wait_for_ready([self.hassette.database_service])
        await self.hassette.database_service.submit(
            self.repository.reconcile_registrations(
                app_key,
                live_listener_ids,
                live_job_ids,
                session_id=session_id,
            )
        )

    async def drain_and_persist(
        self,
        first_item: ExecutionRecord | RetryableBatch | None = None,
    ) -> None:
        """Drain up to 100 queue items and persist them to DB.

        Separates fresh ExecutionRecord items from RetryableBatch items.
        RetryableBatch items are processed separately to preserve their retry_count.

        Note: the 100-item cap applies to *queue items*, not total records.
        A single RetryableBatch counts as 1 queue item but may contain a full
        prior batch's worth of records.  This is acceptable for append-only
        telemetry — a large single transaction at recovery time is benign.

        Args:
            first_item: An already-dequeued item to include as the first record.
                When provided, at most 99 additional items are drained from the queue
                so that the total batch size stays at 100.
        """
        fresh_records: list[ExecutionRecord] = []
        retry_batches: list[RetryableBatch] = []

        def _classify(item: ExecutionRecord | RetryableBatch) -> None:
            if isinstance(item, RetryableBatch):
                retry_batches.append(item)
            elif isinstance(item, ExecutionRecord):
                fresh_records.append(item)
            else:
                typing.assert_never(item)

        if first_item is not None:
            _classify(first_item)

        # Drain remaining items up to a total batch size of _BATCH_DRAIN_CAP (non-blocking)
        for _ in range(_BATCH_DRAIN_CAP - 1 if first_item is not None else _BATCH_DRAIN_CAP):
            try:
                item = self._write_queue.get_nowait()
            except asyncio.QueueEmpty:
                break
            _classify(item)

        # Persist fresh records as a single batch (retry_count=0)
        if fresh_records:
            await self.persist_batch(fresh_records)

        # Process each RetryableBatch separately to preserve its retry_count.
        # Skip batches whose backoff window has not yet elapsed — re-enqueue them.
        now = time.monotonic()
        for batch in retry_batches:
            if batch.not_before > now:
                # Backoff window still active — put it back for a later drain cycle
                try:
                    self._write_queue.put_nowait(batch)
                except asyncio.QueueFull:
                    drop_count = len(batch.records)
                    self._dropped_overflow += drop_count
                    self.logger.error(
                        "Write queue full while deferring retry batch (not_before not reached) "
                        "— dropping %d records (total overflow: %d)",
                        drop_count,
                        self._dropped_overflow,
                    )
                continue
            await self.persist_batch(batch.records, retry_count=batch.retry_count)

    async def flush_queue(self) -> None:
        """Drain and persist ALL remaining items in the write queue.

        Called during shutdown to ensure no records are lost.
        Unlike drain_and_persist, there is no size limit.

        Wraps persist_batch in try/except — DB may already be closed at shutdown.
        """
        records: list[ExecutionRecord] = []

        while True:
            try:
                item = self._write_queue.get_nowait()
            except asyncio.QueueEmpty:
                break

            if isinstance(item, RetryableBatch):
                # retry_count and not_before intentionally bypassed — during shutdown,
                # we make a single best-effort persist regardless of backoff state.
                records.extend(item.records)
            elif isinstance(item, ExecutionRecord):
                records.append(item)
            else:
                typing.assert_never(item)

        if not records:
            return

        try:
            await self.persist_batch(records)
        except Exception:
            drop_count = len(records)
            self._dropped_shutdown += drop_count
            self.logger.error(
                "flush_queue: failed to persist %d records during shutdown — dropped (total shutdown: %d)",
                drop_count,
                self._dropped_shutdown,
            )

    async def persist_batch(
        self,
        records: list[ExecutionRecord],
        *,
        retry_count: int = 0,
    ) -> None:
        """Write a batch of unified execution records to the DB in a single transaction.

        Session injection:
        - Records with session_id=None are updated to the current session_id at drain time.
        - Records with no session available are dropped with a warning.

        Error classification:
        - sqlite3.OperationalError → retry via RetryableBatch (max 3 retries).
        - sqlite3.IntegrityError → FK violation path (row-by-row fallback).
        - sqlite3.DataError / sqlite3.ProgrammingError → non-retryable, drop + REGRESSION log.
        - Other Exception → non-retryable, drop + ERROR log.

        Args:
            records: Unified execution records to insert into executions.
            retry_count: The number of times this batch has already been retried.
        """
        # Drain-time session_id injection
        # Records enqueued before session creation have session_id=None.
        # Inject the real session_id now at persist time.
        current_session_id = self.hassette.try_session_id()

        if current_session_id is not None:
            records = [
                dataclass_replace(r, session_id=current_session_id) if r.session_id is None else r for r in records
            ]
        else:
            # Session still not ready — drop records with None session_id
            no_session = [r for r in records if r.session_id is None]
            if no_session:
                self.logger.warning(
                    "Session not yet created at drain time — dropping %d record(s) with no session_id",
                    len(no_session),
                )
            records = [r for r in records if r.session_id is not None]

        if not records:
            return

        try:
            await self.hassette.database_service.submit(self.repository.persist_execution_batch(records))
            await self.emit_completion_events(records)
        except sqlite3.OperationalError as exc:
            # Retryable — transient DB error (disk I/O, locked, etc.)
            if retry_count >= _MAX_RETRY_COUNT:
                drop_count = len(records)
                self._dropped_exhausted += drop_count
                self.logger.error(
                    "Max retries (%d) exceeded for %d record(s) — dropping (total exhausted: %d): %s",
                    _MAX_RETRY_COUNT,
                    drop_count,
                    self._dropped_exhausted,
                    exc,
                )
            else:
                self.logger.warning(
                    "OperationalError persisting batch — re-enqueueing as RetryableBatch (attempt %d/%d): %s",
                    retry_count + 1,
                    _MAX_RETRY_COUNT,
                    exc,
                )
                try:
                    await asyncio.sleep(0)  # yield event loop before retry to avoid starving fresh records
                    self._write_queue.put_nowait(
                        RetryableBatch(
                            records=list(records),
                            retry_count=retry_count + 1,
                            not_before=time.monotonic() + _RETRY_BACKOFF_BASE_SECONDS * (retry_count + 1),
                        )
                    )
                except asyncio.QueueFull:
                    drop_count = len(records)
                    self._dropped_exhausted += drop_count
                    self.logger.error(
                        "Write queue full while re-enqueueing retry batch — dropping %d records (total exhausted: %d)",
                        drop_count,
                        self._dropped_exhausted,
                    )

        except sqlite3.IntegrityError:
            # FK violation — fall back to row-by-row INSERT
            await self.handle_fk_violation(records)

        except (sqlite3.DataError, sqlite3.ProgrammingError) as exc:
            # Non-retryable schema/data mismatch — this is a regression
            drop_count = len(records)
            self.logger.error(
                "REGRESSION: Non-retryable DB error (%s) — dropping %d record(s): %s",
                type(exc).__name__,
                drop_count,
                exc,
            )

        except Exception as exc:
            # Unknown error — drop and log at ERROR
            drop_count = len(records)
            self.logger.error(
                "Unexpected error persisting %d telemetry record(s) — dropping: %s",
                drop_count,
                exc,
            )

    async def emit_completion_events(
        self,
        records: list[ExecutionRecord],
    ) -> None:
        """Emit bus topic events for persisted execution records.

        Fires ``HASSETTE_EVENT_EXECUTION_COMPLETED`` for each app-tier execution
        (both handler and job kinds). The payload's ``kind`` field distinguishes
        handler from job completions.

        Payloads include ``app_key`` and ``instance_index`` sourced directly from the
        in-memory record (populated at build time from the Listener/ScheduledJob object).

        Errors are suppressed so that emission failures never affect telemetry persistence.
        """
        try:
            app_records = [r for r in records if r.source_tier == "app"]
            # Regression guard: an app-tier completion should always carry an owner.
            # An empty app_key means registration misfired.
            unowned = sum(1 for r in app_records if not r.app_key)
            if unowned:
                self.logger.warning(
                    "Emitting %d app-tier completion event(s) with empty app_key — telemetry will be unattributed",
                    unowned,
                )
            for record in app_records:
                exec_event = HassetteExecutionCompletedEvent.from_record(
                    kind=record.kind,
                    status=record.status,
                    duration_ms=record.duration_ms,
                    listener_id=record.listener_id,
                    job_id=record.job_id,
                    app_key=record.app_key,
                    instance_index=record.instance_index,
                    error_type=record.error_type,
                    thread_leaked=record.thread_leaked,
                )
                await self.hassette.send_event(exec_event)
        except Exception:
            self.logger.debug("Failed to emit completion events — ignoring", exc_info=True)

    def record_blocking_event(self, event: WatchdogEvent | MonkeypatchEvent) -> None:
        """Persist a blocking event row. Must be called on the loop thread.

        Tier 1 (watchdog): the daemon thread marshals via ``loop.call_soon_threadsafe``
        so this method always runs on the loop thread regardless of which tier detected
        the event. Tier 2 (monkeypatch) calls this directly from the wrapper, which
        already executes on the loop thread.

        Enqueues the DB write fire-and-forget via ``database_service.enqueue()``, which
        returns immediately and drops on a full write queue rather than accumulating
        suspended tasks under a detection storm.

        Args:
            event: A ``WatchdogEvent`` (tier='watchdog') or ``MonkeypatchEvent``
                (tier='monkeypatch') describing the detected blocking call.
        """
        session_id = self.hassette.try_session_id()

        if isinstance(event, WatchdogEvent):
            blocking_event = BlockingEvent(
                session_id=session_id,
                app_key=event.app_key,
                instance_name=event.instance_name,
                instance_index=event.instance_index,
                execution_id=event.execution_id,
                tier="watchdog",
                primitive=None,
                # Store the stack text in source_location (Tier 1 has no call-site location,
                # but the captured stack is the closest equivalent).
                source_location=event.stack_text,
                stall_duration_ms=event.stall_duration_ms,
                detected_ts=event.detected_at,
                source_tier="app" if event.app_key is not None else "framework",
                reason=event.reason,
            )
        else:
            blocking_event = BlockingEvent(
                session_id=session_id,
                app_key=event.app_key,
                instance_name=event.instance_name,
                instance_index=event.instance_index,
                execution_id=event.execution_id,
                tier="monkeypatch",
                primitive=event.primitive,
                source_location=event.source_location,
                stall_duration_ms=None,
                detected_ts=event.detected_at,
                source_tier="app" if event.app_key is not None else "framework",
                reason=event.reason,
            )

        # Fire-and-forget telemetry: enqueue() drops on a full write queue (and logs the
        # drop) rather than suspending an unbounded number of spawned tasks under a Tier 2
        # detection storm. Blocking-event rows are diagnostic, not a completion contract.
        #
        # enqueue() raises RuntimeError if the database service queue isn't live yet (before
        # on_initialize) or after shutdown. The Tier 2 monkeypatch guard can fire in exactly
        # those windows — and it runs inside the wrapped primitive (e.g. socket.send), where a
        # raised exception would crash the caller. The detection path is observational, so a
        # not-yet-ready queue drops the row instead of propagating.
        try:
            self.hassette.database_service.enqueue(self.repository.insert_blocking_event(blocking_event))
        except RuntimeError:
            # enqueue() is the only RuntimeError source here, raised when the queue isn't live.
            self.logger.debug("Database service not ready — dropping blocking-event row")

    async def handle_fk_violation(
        self,
        records: list[ExecutionRecord],
    ) -> None:
        """Handle an IntegrityError by re-inserting records with FK fallback.

        Uses a single database_service.submit() call (one queue slot, one
        transaction) to process all records row-by-row. For each record that
        fails with an IntegrityError, the FK field is nulled and retried.

        Args:
            records: Unified execution records to insert individually.
        """
        try:
            dropped = await self.hassette.database_service.submit(
                self.repository.persist_execution_batch_with_fk_fallback(records)
            )
            if dropped > 0:
                self._dropped_exhausted += dropped
                self.logger.error(
                    "FK violation fallback: %d record(s) dropped even with null FK (total exhausted: %d)",
                    dropped,
                    self._dropped_exhausted,
                )
            else:
                await self.emit_completion_events(records)
        except Exception as exc:
            drop_count = len(records)
            self._dropped_exhausted += drop_count
            self.logger.error(
                "FK violation fallback failed entirely — dropping %d record(s) (total exhausted: %d): %s",
                drop_count,
                self._dropped_exhausted,
                exc,
            )

current_execution: ExecutionMarker | None = None class-attribute instance-attribute

Thread-visible marker of the execution currently on the loop thread, or None when idle.

Read by the off-loop blocking-IO watchdog (Tier 1) to attribute a loop freeze to the app that caused it. Published via atomic attribute assignment in bind_execution_context and cleared in unbind_execution_context. Each assignment creates an instance-level attribute that shadows this class-level default, so instances never share marker state. The class-level None default exists only so the attribute is safe to read before the first execution and on instances built via __new__ in test helpers (which bypass __init__ — see tests/unit/core/conftest.py).

repository: TelemetryRepository = TelemetryRepository(hassette.database_service) instance-attribute

Repository for all telemetry SQL writes.

serve() -> None async

Drain the write queue in batches until shutdown, then flush remaining records.

Uses asyncio.wait() with a timeout equal to max_flush_interval_seconds so that records never sit in the queue longer than that interval, even if the batch size threshold is not reached. When the timeout fires (done is empty), whatever is currently in the queue is drained immediately.

Source code in src/hassette/core/command_executor.py

async def serve(self) -> None:
    """Drain the write queue in batches until shutdown, then flush remaining records.

    Uses asyncio.wait() with a timeout equal to max_flush_interval_seconds so that records
    never sit in the queue longer than that interval, even if the batch size
    threshold is not reached.  When the timeout fires (done is empty), whatever
    is currently in the queue is drained immediately.
    """
    mark_ready(self, reason="CommandExecutor started")
    flush_interval = self.hassette.config.database.max_flush_interval_seconds

    while True:
        get_fut = asyncio.create_task(self._write_queue.get())
        shutdown_fut = asyncio.create_task(self.shutdown_event.wait())

        done, pending = await asyncio.wait(
            [get_fut, shutdown_fut],
            timeout=flush_interval,
            return_when=asyncio.FIRST_COMPLETED,
        )

        # Cancel the pending futures to avoid task leaks
        for fut in pending:
            fut.cancel()
            with contextlib.suppress(asyncio.CancelledError, Exception):
                await fut

        if self.shutdown_event.is_set():
            # get_fut dequeued an item before shutdown was detected — re-enqueue so flush_queue() picks it up
            if get_fut in done and not get_fut.cancelled() and get_fut.exception() is None:
                result = get_fut.result()
                try:
                    self._write_queue.put_nowait(result)
                except asyncio.QueueFull:
                    self._dropped_overflow += 1
                    self.logger.error(
                        "Write queue full during shutdown — dropping 1 record (total dropped: %d)",
                        self._dropped_overflow,
                    )
            await self.flush_queue()
            return

        if get_fut in done and not get_fut.cancelled() and get_fut.exception() is None:
            # An item arrived — drain the full queue in one batch
            try:
                await self.drain_and_persist(first_item=get_fut.result())
            except Exception:
                self.logger.exception(
                    "drain_and_persist failed — records from this batch are dropped (already dequeued)"
                )
        elif not done:
            # Timeout — timer fired; drain whatever accumulated without a triggering item
            if not self._write_queue.empty():
                try:
                    await self.drain_and_persist()
                except Exception:
                    self.logger.exception(
                        "drain_and_persist failed (timer flush) — records from this batch may be dropped"
                    )

get_drop_counters() -> tuple[int, int, int]

Return (dropped_overflow, dropped_exhausted, dropped_shutdown) counters.

Returns:

Type	Description
int	A tuple of counters where:
int	overflow_count: records dropped because the write queue was full.
int	exhausted_count: records dropped because max retries were exceeded.
tuple[int, int, int]	shutdown_count: records dropped during shutdown flush.

Source code in src/hassette/core/command_executor.py

def get_drop_counters(self) -> tuple[int, int, int]:
    """Return (dropped_overflow, dropped_exhausted, dropped_shutdown) counters.

    Returns:
        A tuple of counters where:
        - overflow_count: records dropped because the write queue was full.
        - exhausted_count: records dropped because max retries were exceeded.
        - shutdown_count: records dropped during shutdown flush.
    """
    return (self._dropped_overflow, self._dropped_exhausted, self._dropped_shutdown)

get_error_handler_failures() -> int

Return the count of user error handler invocations that raised or timed out.

Incremented each time a user-registered error handler (bus or scheduler) raises an exception or times out during invocation.

Returns:

Type	Description
int	The cumulative error handler failure count for this session.

Source code in src/hassette/core/command_executor.py

def get_error_handler_failures(self) -> int:
    """Return the count of user error handler invocations that raised or timed out.

    Incremented each time a user-registered error handler (bus or scheduler)
    raises an exception or times out during invocation.

    Returns:
        The cumulative error handler failure count for this session.
    """
    return self._error_handler_failures

execute(cmd: InvokeHandler | ExecuteJob) -> None async

Execute a command (handler invocation or scheduled job).

Parameters:

Name	Type	Description	Default
cmd	InvokeHandler | ExecuteJob	The command to execute.	required

Source code in src/hassette/core/command_executor.py

async def execute(self, cmd: InvokeHandler | ExecuteJob) -> None:
    """Execute a command (handler invocation or scheduled job).

    Args:
        cmd: The command to execute.
    """
    match cmd:
        case InvokeHandler():
            await self.execute_handler(cmd)
        case ExecuteJob():
            await self.execute_job(cmd)

log_timeout_rate_limited(cmd: InvokeHandler | ExecuteJob, result: ExecutionResult) -> None

Log a timeout WARNING, rate-limited per entity (60s suppression window).

Uses the in-memory ID (listener_id for handlers, object identity for jobs) to key the suppression window. Lazily evicts stale entries (>60s old) during each check.

Source code in src/hassette/core/command_executor.py

def log_timeout_rate_limited(self, cmd: InvokeHandler | ExecuteJob, result: ExecutionResult) -> None:
    """Log a timeout WARNING, rate-limited per entity (60s suppression window).

    Uses the in-memory ID (``listener_id`` for handlers, object identity for jobs)
    to key the suppression window. Lazily evicts stale entries (>60s old)
    during each check.
    """
    now = time.monotonic()

    # Determine the in-memory ID for rate-limiting
    match cmd:
        case InvokeHandler():
            entity_id = cmd.listener.listener_id
            label = f"listener_id={cmd.listener.listener_id}, topic={cmd.topic}"
        case ExecuteJob():
            entity_id = id(cmd.job)
            label = f"job_db_id={cmd.job_db_id}, name={cmd.job.name}"

    # Lazy eviction of stale entries, then cap to bound memory under sustained unavailability
    stale_ids = [k for k, ts in self._timeout_warn_timestamps.items() if now - ts > _TIMEOUT_WARN_SUPPRESS_SECS]
    for k in stale_ids:
        del self._timeout_warn_timestamps[k]
    if len(self._timeout_warn_timestamps) > _TIMEOUT_WARN_CACHE_MAX:
        self._timeout_warn_timestamps.clear()

    # Rate-limit check
    last_ts = self._timeout_warn_timestamps.get(entity_id)
    if last_ts is not None and now - last_ts < _TIMEOUT_WARN_SUPPRESS_SECS:
        return  # suppressed
    self._timeout_warn_timestamps[entity_id] = now

    self.logger.warning(
        "Execution timed out after %.1fms (%s, timeout=%.1fs)",
        result.duration_ms,
        label,
        cmd.effective_timeout,
    )

enqueue_record(record: ExecutionRecord) -> None

Enqueue a record, dropping and logging if the queue is full.

Also logs a WARNING when the queue exceeds 75% capacity (rate-limited).

Source code in src/hassette/core/command_executor.py

def enqueue_record(self, record: ExecutionRecord) -> None:
    """Enqueue a record, dropping and logging if the queue is full.

    Also logs a WARNING when the queue exceeds 75% capacity (rate-limited).
    """
    max_size = self._write_queue.maxsize
    current_size = self._write_queue.qsize()

    # 75% capacity warning (rate-limited)
    if max_size > 0 and current_size >= int(max_size * _CAPACITY_WARN_THRESHOLD):
        now = time.monotonic()
        if now - self._last_capacity_warn_ts >= _CAPACITY_WARN_RATE_LIMIT_SECS:
            self._last_capacity_warn_ts = now
            self.logger.warning(
                "Write queue at %d/%d (%.0f%%) — high telemetry load",
                current_size,
                max_size,
                (current_size / max_size) * 100,
            )

    try:
        self._write_queue.put_nowait(record)
    except asyncio.QueueFull:
        self._dropped_overflow += 1
        self.logger.error(
            "Write queue full (%d/%d) — dropping record (total dropped: %d)",
            current_size,
            max_size,
            self._dropped_overflow,
        )

build_record(cmd: InvokeHandler | ExecuteJob, result: ExecutionResult, execution_start_ts: float, execution_id: str) -> ExecutionRecord

Build a unified ExecutionRecord from the execution result and command.

session_id is set to None if the session hasn't been created yet (pre-Phase 1). The actual session_id is injected at drain time in persist_batch.

Parameters:

Name	Type	Description	Default
cmd	InvokeHandler | ExecuteJob	The originating command.	required
result	ExecutionResult	The execution result with timing and error info.	required
execution_start_ts	float	Unix timestamp when execution began.	required
execution_id	str	UUIDv7 string for this execution instance.	required

Source code in src/hassette/core/command_executor.py

def build_record(
    self,
    cmd: InvokeHandler | ExecuteJob,
    result: ExecutionResult,
    execution_start_ts: float,
    execution_id: str,
) -> ExecutionRecord:
    """Build a unified ExecutionRecord from the execution result and command.

    session_id is set to None if the session hasn't been created yet (pre-Phase 1).
    The actual session_id is injected at drain time in persist_batch.

    Args:
        cmd: The originating command.
        result: The execution result with timing and error info.
        execution_start_ts: Unix timestamp when execution began.
        execution_id: UUIDv7 string for this execution instance.
    """
    session_id = self.hassette.try_session_id()

    match cmd:
        case InvokeHandler():
            return ExecutionRecord(
                kind="handler",
                listener_id=cmd.listener_id,
                job_id=None,
                session_id=session_id,
                execution_start_ts=execution_start_ts,
                duration_ms=result.duration_ms,
                status=result.status,
                app_key=cmd.listener.identity.app_key,
                instance_index=cmd.listener.identity.instance_index,
                source_tier=cmd.source_tier,
                is_di_failure=result.is_di_failure,
                thread_leaked=result.thread_leaked,
                error_type=result.error_type,
                error_message=result.error_message,
                error_traceback=result.error_traceback,
                execution_id=execution_id,
                trigger_context_id=None if cmd.is_synthetic else cmd.event.payload.event_id,
                trigger_origin=SYNTHETIC_ORIGIN if cmd.is_synthetic else cmd.event.payload.origin,
            )
        case ExecuteJob():
            return ExecutionRecord(
                kind="job",
                listener_id=None,
                job_id=cmd.job_db_id,
                session_id=session_id,
                execution_start_ts=execution_start_ts,
                duration_ms=result.duration_ms,
                status=result.status,
                app_key=cmd.job.app_key,
                instance_index=cmd.job.instance_index,
                source_tier=cmd.source_tier,
                is_di_failure=result.is_di_failure,
                thread_leaked=result.thread_leaked,
                error_type=result.error_type,
                error_message=result.error_message,
                error_traceback=result.error_traceback,
                execution_id=execution_id,
                trigger_mode=cmd.trigger_mode,
            )

bind_execution_context(app_key: str | None, instance_index: int, instance_name: str | None, *, execution_kind: str | None = None, listener_id: int | None = None, job_id: int | None = None) -> tuple[str, Token[str | None]]

Set CURRENT_EXECUTION_ID and bind structlog context vars for the duration of an execution.

Source code in src/hassette/core/command_executor.py

def bind_execution_context(
    self,
    app_key: str | None,
    instance_index: int,
    instance_name: str | None,
    *,
    execution_kind: str | None = None,
    listener_id: int | None = None,
    job_id: int | None = None,
) -> tuple[str, Token[str | None]]:
    """Set CURRENT_EXECUTION_ID and bind structlog context vars for the duration of an execution."""
    execution_id = str(uuid_utils.uuid7())
    token = CURRENT_EXECUTION_ID.set(execution_id)
    resolved_app_key = app_key or None
    structlog.contextvars.bind_contextvars(
        app_key=resolved_app_key,
        instance_name=instance_name,
        instance_index=instance_index,
        execution_kind=execution_kind,
        listener_id=listener_id,
        job_id=job_id,
    )
    # Capture the owning task identity so a cross-thread reader can confirm this marker
    # names the task actually frozen on the loop, not a displaced one. This runs inside an
    # execute_handler/execute_job task in production; guard the no-running-loop case so a
    # direct synchronous call (tests) does not raise.
    try:
        current_task = asyncio.current_task()
    except RuntimeError:
        current_task = None
    # Publish the thread-visible marker last, as a single atomic assignment, so the off-loop
    # watchdog reads a fully-formed snapshot of the execution now holding the loop thread.
    self.current_execution = ExecutionMarker(
        app_key=resolved_app_key,
        instance_name=instance_name,
        execution_id=execution_id,
        started_at=time.monotonic(),
        instance_index=instance_index,
        task_id=id(current_task) if current_task is not None else None,
    )
    return execution_id, token

unbind_execution_context(token: Token[str | None]) -> None

Clear the execution marker and reset the ContextVar/structlog binding on exit.

Paired with bind_execution_context. Was a @staticmethod; it takes self now so it can clear current_execution. Clearing the marker first (to None = idle) is deliberate: the off-loop watchdog reads only current_execution, so a clear-first order means it can never attribute a stale execution once this returns. CURRENT_EXECUTION_ID is thread-local, so its reset order relative to the marker is invisible to other threads.

Source code in src/hassette/core/command_executor.py

def unbind_execution_context(self, token: Token[str | None]) -> None:
    """Clear the execution marker and reset the ContextVar/structlog binding on exit.

    Paired with ``bind_execution_context``. Was a ``@staticmethod``; it takes ``self`` now
    so it can clear ``current_execution``. Clearing the marker first (to ``None`` = idle) is
    deliberate: the off-loop watchdog reads only ``current_execution``, so a clear-first order
    means it can never attribute a stale execution once this returns. ``CURRENT_EXECUTION_ID``
    is thread-local, so its reset order relative to the marker is invisible to other threads.
    """
    self.current_execution = None
    CURRENT_EXECUTION_ID.reset(token)
    structlog.contextvars.unbind_contextvars(
        "app_key", "instance_name", "instance_index", "execution_kind", "listener_id", "job_id"
    )

execute_handler(cmd: InvokeHandler) -> None async

Execute a listener handler invocation and queue the result record.

Source code in src/hassette/core/command_executor.py

async def execute_handler(self, cmd: InvokeHandler) -> None:
    """Execute a listener handler invocation and queue the result record."""
    execution_id, token = self.bind_execution_context(
        cmd.listener.identity.app_key,
        cmd.listener.identity.instance_index,
        cmd.listener.identity.instance_name,
        execution_kind="handler",
        listener_id=cmd.listener_id,
    )
    try:

        def log_error(result: ExecutionResult) -> None:
            if result.error_traceback is None:
                self.logger.error(
                    "Handler error (topic=%s, exec=%s): %s", cmd.topic, execution_id, result.error_message
                )
            else:
                self.logger.error(
                    "Handler error (topic=%s, handler=%r, exec=%s)\n%s",
                    cmd.topic,
                    cmd.listener,
                    execution_id,
                    result.error_traceback,
                )

        result = await self._execute(lambda: cmd.listener.invoker.invoke(cmd.event), cmd, log_error, execution_id)

        if (result.is_error or result.is_timed_out) and result.exc is not None:
            error_handler = cmd.listener.invoker.error_handler or cmd.app_level_error_handler
            if error_handler is not None:
                ctx = BusErrorContext(
                    exception=result.exc,
                    traceback="".join(traceback.format_exception(result.exc)),
                    execution_id=execution_id,
                    topic=cmd.topic,
                    listener_name=repr(cmd.listener),
                    event=cmd.event,
                )
                # FIXME(#573): no per-listener rate-limit on error handler spawns — high-frequency
                # failures can accumulate unbounded concurrent tasks.
                self.task_bucket.spawn(
                    self.invoke_error_handler(error_handler, ctx),
                    name="executor:bus_error_handler",
                )
    finally:
        self.unbind_execution_context(token)

execute_job(cmd: ExecuteJob) -> None async

Execute a scheduled job and queue the result record.

Source code in src/hassette/core/command_executor.py

async def execute_job(self, cmd: ExecuteJob) -> None:
    """Execute a scheduled job and queue the result record."""
    execution_id, token = self.bind_execution_context(
        cmd.job.app_key,
        cmd.job.instance_index,
        cmd.job.instance_name,
        execution_kind="job",
        job_id=cmd.job_db_id,
    )
    try:

        def log_error(result: ExecutionResult) -> None:
            if result.error_traceback is None:
                self.logger.error(
                    "Job error (job_db_id=%s, exec=%s): %s", cmd.job_db_id, execution_id, result.error_message
                )
            else:
                self.logger.error(
                    "Job error (job_db_id=%s, exec=%s)\n%s", cmd.job_db_id, execution_id, result.error_traceback
                )

        result = await self._execute(cmd.callable, cmd, log_error, execution_id)

        if (result.is_error or result.is_timed_out) and result.exc is not None:
            error_handler = cmd.job.error_handler or cmd.app_level_error_handler
            if error_handler is not None:
                ctx = SchedulerErrorContext(
                    exception=result.exc,
                    traceback="".join(traceback.format_exception(result.exc)),
                    execution_id=execution_id,
                    job_name=cmd.job.name,
                    job_group=cmd.job.group,
                    args=cmd.job.args,
                    kwargs=dict(cmd.job.kwargs),
                )
                # FIXME(#573): no per-job rate-limit on error handler spawns.
                self.task_bucket.spawn(
                    self.invoke_error_handler(error_handler, ctx),
                    name="executor:scheduler_error_handler",
                )
    finally:
        self.unbind_execution_context(token)

invoke_error_handler(handler: Callable, ctx: ErrorContext) -> None async

Invoke a user-registered error handler in a separate spawned task.

Note: CURRENT_EXECUTION_ID is set to the parent execution's ID via inherited context snapshot. This is intentional for error correlation but is not reset here — sub-tasks spawned by user error handler code will inherit the same ID.

Source code in src/hassette/core/command_executor.py

async def invoke_error_handler(
    self,
    handler: "Callable",
    ctx: ErrorContext,
) -> None:
    """Invoke a user-registered error handler in a separate spawned task.

    Note: CURRENT_EXECUTION_ID is set to the parent execution's ID via inherited
    context snapshot. This is intentional for error correlation but is not reset
    here — sub-tasks spawned by user error handler code will inherit the same ID.
    """
    async_handler = self.task_bucket.make_async_adapter(handler)
    timeout = self.hassette.config.lifecycle.error_handler_timeout_seconds
    label = ctx.log_label
    try:
        async with asyncio.timeout(timeout):
            await async_handler(ctx)
    except TimeoutError:
        self._error_handler_failures += 1
        if timeout is None:
            self.logger.exception("Error handler raised TimeoutError (%s)", label)
        else:
            self.logger.warning("Error handler timed out after %.1fs (%s)", timeout, label)
    except Exception:
        self._error_handler_failures += 1
        self.logger.exception("Error handler raised an exception (%s)", label)

register_listener(registration: ListenerRegistration) -> int async

Insert a listener registration into the listeners table.

Parameters:

Name	Type	Description	Default
registration	ListenerRegistration	The listener registration data.	required

Returns:

Type	Description
int	The row ID of the inserted row.

Source code in src/hassette/core/command_executor.py

async def register_listener(self, registration: ListenerRegistration) -> int:
    """Insert a listener registration into the listeners table.

    Args:
        registration: The listener registration data.

    Returns:
        The row ID of the inserted row.
    """
    listener_id = await self.hassette.database_service.submit(self.repository.register_listener(registration))
    return listener_id

register_job(registration: ScheduledJobRegistration) -> int async

Insert a scheduled job registration into the scheduled_jobs table.

Parameters:

Name	Type	Description	Default
registration	ScheduledJobRegistration	The scheduled job registration data.	required

Returns:

Type	Description
int	The row ID of the inserted row.

Source code in src/hassette/core/command_executor.py

async def register_job(self, registration: ScheduledJobRegistration) -> int:
    """Insert a scheduled job registration into the scheduled_jobs table.

    Args:
        registration: The scheduled job registration data.

    Returns:
        The row ID of the inserted row.
    """
    job_id = await self.hassette.database_service.submit(self.repository.register_job(registration))
    return job_id

mark_job_cancelled(db_id: int) -> None async

Set cancelled_at on the scheduled_jobs row to persist durable cancellation state.

Delegates to TelemetryRepository.mark_job_cancelled via DatabaseService.submit.

Parameters:

Name	Type	Description	Default
db_id	int	The id of the scheduled_jobs row to mark as cancelled.	required

Source code in src/hassette/core/command_executor.py

async def mark_job_cancelled(self, db_id: int) -> None:
    """Set ``cancelled_at`` on the scheduled_jobs row to persist durable cancellation state.

    Delegates to ``TelemetryRepository.mark_job_cancelled`` via ``DatabaseService.submit``.

    Args:
        db_id: The ``id`` of the ``scheduled_jobs`` row to mark as cancelled.
    """
    await self.hassette.database_service.submit(self.repository.mark_job_cancelled(db_id))

mark_listener_cancelled(db_id: int) -> None async

Set cancelled_at on the listeners row to persist durable cancellation state.

Delegates to TelemetryRepository.mark_listener_cancelled via DatabaseService.submit.

Parameters:

Name	Type	Description	Default
db_id	int	The id of the listeners row to mark as cancelled.	required

Source code in src/hassette/core/command_executor.py

async def mark_listener_cancelled(self, db_id: int) -> None:
    """Set ``cancelled_at`` on the listeners row to persist durable cancellation state.

    Delegates to ``TelemetryRepository.mark_listener_cancelled`` via ``DatabaseService.submit``.

    Args:
        db_id: The ``id`` of the ``listeners`` row to mark as cancelled.
    """
    await self.hassette.database_service.submit(self.repository.mark_listener_cancelled(db_id))

reconcile_registrations(app_key: str, live_listener_ids: list[int], live_job_ids: list[int], *, session_id: int | None = None) -> None async

Reconcile listener and job registrations for an app after initialization.

Deletes stale rows without history, sets retired_at on stale rows with history, and deletes once=True rows from previous sessions. Delegates to TelemetryRepository.reconcile_registrations via DatabaseService.submit.

Parameters:

Name	Type	Description	Default
app_key	str	The app key to reconcile.	required
live_listener_ids	list[int]	IDs of currently active listener rows.	required
live_job_ids	list[int]	IDs of currently active scheduled_job rows.	required
session_id	int | None	Current session ID, used to guard once=True row deletion.	None

Source code in src/hassette/core/command_executor.py

async def reconcile_registrations(
    self,
    app_key: str,
    live_listener_ids: list[int],
    live_job_ids: list[int],
    *,
    session_id: int | None = None,
) -> None:
    """Reconcile listener and job registrations for an app after initialization.

    Deletes stale rows without history, sets ``retired_at`` on stale rows with
    history, and deletes ``once=True`` rows from previous sessions. Delegates
    to ``TelemetryRepository.reconcile_registrations`` via ``DatabaseService.submit``.

    Args:
        app_key: The app key to reconcile.
        live_listener_ids: IDs of currently active listener rows.
        live_job_ids: IDs of currently active scheduled_job rows.
        session_id: Current session ID, used to guard once=True row deletion.
    """
    await self.hassette.wait_for_ready([self.hassette.database_service])
    await self.hassette.database_service.submit(
        self.repository.reconcile_registrations(
            app_key,
            live_listener_ids,
            live_job_ids,
            session_id=session_id,
        )
    )

drain_and_persist(first_item: ExecutionRecord | RetryableBatch | None = None) -> None async

Drain up to 100 queue items and persist them to DB.

Separates fresh ExecutionRecord items from RetryableBatch items. RetryableBatch items are processed separately to preserve their retry_count.

Note: the 100-item cap applies to queue items, not total records. A single RetryableBatch counts as 1 queue item but may contain a full prior batch's worth of records. This is acceptable for append-only telemetry — a large single transaction at recovery time is benign.

Parameters:

Name	Type	Description	Default
first_item	ExecutionRecord | RetryableBatch | None	An already-dequeued item to include as the first record. When provided, at most 99 additional items are drained from the queue so that the total batch size stays at 100.	None

Source code in src/hassette/core/command_executor.py

async def drain_and_persist(
    self,
    first_item: ExecutionRecord | RetryableBatch | None = None,
) -> None:
    """Drain up to 100 queue items and persist them to DB.

    Separates fresh ExecutionRecord items from RetryableBatch items.
    RetryableBatch items are processed separately to preserve their retry_count.

    Note: the 100-item cap applies to *queue items*, not total records.
    A single RetryableBatch counts as 1 queue item but may contain a full
    prior batch's worth of records.  This is acceptable for append-only
    telemetry — a large single transaction at recovery time is benign.

    Args:
        first_item: An already-dequeued item to include as the first record.
            When provided, at most 99 additional items are drained from the queue
            so that the total batch size stays at 100.
    """
    fresh_records: list[ExecutionRecord] = []
    retry_batches: list[RetryableBatch] = []

    def _classify(item: ExecutionRecord | RetryableBatch) -> None:
        if isinstance(item, RetryableBatch):
            retry_batches.append(item)
        elif isinstance(item, ExecutionRecord):
            fresh_records.append(item)
        else:
            typing.assert_never(item)

    if first_item is not None:
        _classify(first_item)

    # Drain remaining items up to a total batch size of _BATCH_DRAIN_CAP (non-blocking)
    for _ in range(_BATCH_DRAIN_CAP - 1 if first_item is not None else _BATCH_DRAIN_CAP):
        try:
            item = self._write_queue.get_nowait()
        except asyncio.QueueEmpty:
            break
        _classify(item)

    # Persist fresh records as a single batch (retry_count=0)
    if fresh_records:
        await self.persist_batch(fresh_records)

    # Process each RetryableBatch separately to preserve its retry_count.
    # Skip batches whose backoff window has not yet elapsed — re-enqueue them.
    now = time.monotonic()
    for batch in retry_batches:
        if batch.not_before > now:
            # Backoff window still active — put it back for a later drain cycle
            try:
                self._write_queue.put_nowait(batch)
            except asyncio.QueueFull:
                drop_count = len(batch.records)
                self._dropped_overflow += drop_count
                self.logger.error(
                    "Write queue full while deferring retry batch (not_before not reached) "
                    "— dropping %d records (total overflow: %d)",
                    drop_count,
                    self._dropped_overflow,
                )
            continue
        await self.persist_batch(batch.records, retry_count=batch.retry_count)

flush_queue() -> None async

Drain and persist ALL remaining items in the write queue.

Called during shutdown to ensure no records are lost. Unlike drain_and_persist, there is no size limit.

Wraps persist_batch in try/except — DB may already be closed at shutdown.

Source code in src/hassette/core/command_executor.py

async def flush_queue(self) -> None:
    """Drain and persist ALL remaining items in the write queue.

    Called during shutdown to ensure no records are lost.
    Unlike drain_and_persist, there is no size limit.

    Wraps persist_batch in try/except — DB may already be closed at shutdown.
    """
    records: list[ExecutionRecord] = []

    while True:
        try:
            item = self._write_queue.get_nowait()
        except asyncio.QueueEmpty:
            break

        if isinstance(item, RetryableBatch):
            # retry_count and not_before intentionally bypassed — during shutdown,
            # we make a single best-effort persist regardless of backoff state.
            records.extend(item.records)
        elif isinstance(item, ExecutionRecord):
            records.append(item)
        else:
            typing.assert_never(item)

    if not records:
        return

    try:
        await self.persist_batch(records)
    except Exception:
        drop_count = len(records)
        self._dropped_shutdown += drop_count
        self.logger.error(
            "flush_queue: failed to persist %d records during shutdown — dropped (total shutdown: %d)",
            drop_count,
            self._dropped_shutdown,
        )

persist_batch(records: list[ExecutionRecord], *, retry_count: int = 0) -> None async

Write a batch of unified execution records to the DB in a single transaction.

Session injection: - Records with session_id=None are updated to the current session_id at drain time. - Records with no session available are dropped with a warning.

Error classification: - sqlite3.OperationalError → retry via RetryableBatch (max 3 retries). - sqlite3.IntegrityError → FK violation path (row-by-row fallback). - sqlite3.DataError / sqlite3.ProgrammingError → non-retryable, drop + REGRESSION log. - Other Exception → non-retryable, drop + ERROR log.

Parameters:

Name	Type	Description	Default
records	list[ExecutionRecord]	Unified execution records to insert into executions.	required
retry_count	int	The number of times this batch has already been retried.	0

Source code in src/hassette/core/command_executor.py

async def persist_batch(
    self,
    records: list[ExecutionRecord],
    *,
    retry_count: int = 0,
) -> None:
    """Write a batch of unified execution records to the DB in a single transaction.

    Session injection:
    - Records with session_id=None are updated to the current session_id at drain time.
    - Records with no session available are dropped with a warning.

    Error classification:
    - sqlite3.OperationalError → retry via RetryableBatch (max 3 retries).
    - sqlite3.IntegrityError → FK violation path (row-by-row fallback).
    - sqlite3.DataError / sqlite3.ProgrammingError → non-retryable, drop + REGRESSION log.
    - Other Exception → non-retryable, drop + ERROR log.

    Args:
        records: Unified execution records to insert into executions.
        retry_count: The number of times this batch has already been retried.
    """
    # Drain-time session_id injection
    # Records enqueued before session creation have session_id=None.
    # Inject the real session_id now at persist time.
    current_session_id = self.hassette.try_session_id()

    if current_session_id is not None:
        records = [
            dataclass_replace(r, session_id=current_session_id) if r.session_id is None else r for r in records
        ]
    else:
        # Session still not ready — drop records with None session_id
        no_session = [r for r in records if r.session_id is None]
        if no_session:
            self.logger.warning(
                "Session not yet created at drain time — dropping %d record(s) with no session_id",
                len(no_session),
            )
        records = [r for r in records if r.session_id is not None]

    if not records:
        return

    try:
        await self.hassette.database_service.submit(self.repository.persist_execution_batch(records))
        await self.emit_completion_events(records)
    except sqlite3.OperationalError as exc:
        # Retryable — transient DB error (disk I/O, locked, etc.)
        if retry_count >= _MAX_RETRY_COUNT:
            drop_count = len(records)
            self._dropped_exhausted += drop_count
            self.logger.error(
                "Max retries (%d) exceeded for %d record(s) — dropping (total exhausted: %d): %s",
                _MAX_RETRY_COUNT,
                drop_count,
                self._dropped_exhausted,
                exc,
            )
        else:
            self.logger.warning(
                "OperationalError persisting batch — re-enqueueing as RetryableBatch (attempt %d/%d): %s",
                retry_count + 1,
                _MAX_RETRY_COUNT,
                exc,
            )
            try:
                await asyncio.sleep(0)  # yield event loop before retry to avoid starving fresh records
                self._write_queue.put_nowait(
                    RetryableBatch(
                        records=list(records),
                        retry_count=retry_count + 1,
                        not_before=time.monotonic() + _RETRY_BACKOFF_BASE_SECONDS * (retry_count + 1),
                    )
                )
            except asyncio.QueueFull:
                drop_count = len(records)
                self._dropped_exhausted += drop_count
                self.logger.error(
                    "Write queue full while re-enqueueing retry batch — dropping %d records (total exhausted: %d)",
                    drop_count,
                    self._dropped_exhausted,
                )

    except sqlite3.IntegrityError:
        # FK violation — fall back to row-by-row INSERT
        await self.handle_fk_violation(records)

    except (sqlite3.DataError, sqlite3.ProgrammingError) as exc:
        # Non-retryable schema/data mismatch — this is a regression
        drop_count = len(records)
        self.logger.error(
            "REGRESSION: Non-retryable DB error (%s) — dropping %d record(s): %s",
            type(exc).__name__,
            drop_count,
            exc,
        )

    except Exception as exc:
        # Unknown error — drop and log at ERROR
        drop_count = len(records)
        self.logger.error(
            "Unexpected error persisting %d telemetry record(s) — dropping: %s",
            drop_count,
            exc,
        )

emit_completion_events(records: list[ExecutionRecord]) -> None async

Emit bus topic events for persisted execution records.

Fires HASSETTE_EVENT_EXECUTION_COMPLETED for each app-tier execution (both handler and job kinds). The payload's kind field distinguishes handler from job completions.

Payloads include app_key and instance_index sourced directly from the in-memory record (populated at build time from the Listener/ScheduledJob object).

Errors are suppressed so that emission failures never affect telemetry persistence.

Source code in src/hassette/core/command_executor.py

async def emit_completion_events(
    self,
    records: list[ExecutionRecord],
) -> None:
    """Emit bus topic events for persisted execution records.

    Fires ``HASSETTE_EVENT_EXECUTION_COMPLETED`` for each app-tier execution
    (both handler and job kinds). The payload's ``kind`` field distinguishes
    handler from job completions.

    Payloads include ``app_key`` and ``instance_index`` sourced directly from the
    in-memory record (populated at build time from the Listener/ScheduledJob object).

    Errors are suppressed so that emission failures never affect telemetry persistence.
    """
    try:
        app_records = [r for r in records if r.source_tier == "app"]
        # Regression guard: an app-tier completion should always carry an owner.
        # An empty app_key means registration misfired.
        unowned = sum(1 for r in app_records if not r.app_key)
        if unowned:
            self.logger.warning(
                "Emitting %d app-tier completion event(s) with empty app_key — telemetry will be unattributed",
                unowned,
            )
        for record in app_records:
            exec_event = HassetteExecutionCompletedEvent.from_record(
                kind=record.kind,
                status=record.status,
                duration_ms=record.duration_ms,
                listener_id=record.listener_id,
                job_id=record.job_id,
                app_key=record.app_key,
                instance_index=record.instance_index,
                error_type=record.error_type,
                thread_leaked=record.thread_leaked,
            )
            await self.hassette.send_event(exec_event)
    except Exception:
        self.logger.debug("Failed to emit completion events — ignoring", exc_info=True)

record_blocking_event(event: WatchdogEvent | MonkeypatchEvent) -> None

Persist a blocking event row. Must be called on the loop thread.

Tier 1 (watchdog): the daemon thread marshals via loop.call_soon_threadsafe so this method always runs on the loop thread regardless of which tier detected the event. Tier 2 (monkeypatch) calls this directly from the wrapper, which already executes on the loop thread.

Enqueues the DB write fire-and-forget via database_service.enqueue(), which returns immediately and drops on a full write queue rather than accumulating suspended tasks under a detection storm.

Parameters:

Name	Type	Description	Default
event	WatchdogEvent | MonkeypatchEvent	A WatchdogEvent (tier='watchdog') or MonkeypatchEvent (tier='monkeypatch') describing the detected blocking call.	required

Source code in src/hassette/core/command_executor.py

def record_blocking_event(self, event: WatchdogEvent | MonkeypatchEvent) -> None:
    """Persist a blocking event row. Must be called on the loop thread.

    Tier 1 (watchdog): the daemon thread marshals via ``loop.call_soon_threadsafe``
    so this method always runs on the loop thread regardless of which tier detected
    the event. Tier 2 (monkeypatch) calls this directly from the wrapper, which
    already executes on the loop thread.

    Enqueues the DB write fire-and-forget via ``database_service.enqueue()``, which
    returns immediately and drops on a full write queue rather than accumulating
    suspended tasks under a detection storm.

    Args:
        event: A ``WatchdogEvent`` (tier='watchdog') or ``MonkeypatchEvent``
            (tier='monkeypatch') describing the detected blocking call.
    """
    session_id = self.hassette.try_session_id()

    if isinstance(event, WatchdogEvent):
        blocking_event = BlockingEvent(
            session_id=session_id,
            app_key=event.app_key,
            instance_name=event.instance_name,
            instance_index=event.instance_index,
            execution_id=event.execution_id,
            tier="watchdog",
            primitive=None,
            # Store the stack text in source_location (Tier 1 has no call-site location,
            # but the captured stack is the closest equivalent).
            source_location=event.stack_text,
            stall_duration_ms=event.stall_duration_ms,
            detected_ts=event.detected_at,
            source_tier="app" if event.app_key is not None else "framework",
            reason=event.reason,
        )
    else:
        blocking_event = BlockingEvent(
            session_id=session_id,
            app_key=event.app_key,
            instance_name=event.instance_name,
            instance_index=event.instance_index,
            execution_id=event.execution_id,
            tier="monkeypatch",
            primitive=event.primitive,
            source_location=event.source_location,
            stall_duration_ms=None,
            detected_ts=event.detected_at,
            source_tier="app" if event.app_key is not None else "framework",
            reason=event.reason,
        )

    # Fire-and-forget telemetry: enqueue() drops on a full write queue (and logs the
    # drop) rather than suspending an unbounded number of spawned tasks under a Tier 2
    # detection storm. Blocking-event rows are diagnostic, not a completion contract.
    #
    # enqueue() raises RuntimeError if the database service queue isn't live yet (before
    # on_initialize) or after shutdown. The Tier 2 monkeypatch guard can fire in exactly
    # those windows — and it runs inside the wrapped primitive (e.g. socket.send), where a
    # raised exception would crash the caller. The detection path is observational, so a
    # not-yet-ready queue drops the row instead of propagating.
    try:
        self.hassette.database_service.enqueue(self.repository.insert_blocking_event(blocking_event))
    except RuntimeError:
        # enqueue() is the only RuntimeError source here, raised when the queue isn't live.
        self.logger.debug("Database service not ready — dropping blocking-event row")

handle_fk_violation(records: list[ExecutionRecord]) -> None async

Handle an IntegrityError by re-inserting records with FK fallback.

Uses a single database_service.submit() call (one queue slot, one transaction) to process all records row-by-row. For each record that fails with an IntegrityError, the FK field is nulled and retried.

Parameters:

Name	Type	Description	Default
records	list[ExecutionRecord]	Unified execution records to insert individually.	required

Source code in src/hassette/core/command_executor.py

async def handle_fk_violation(
    self,
    records: list[ExecutionRecord],
) -> None:
    """Handle an IntegrityError by re-inserting records with FK fallback.

    Uses a single database_service.submit() call (one queue slot, one
    transaction) to process all records row-by-row. For each record that
    fails with an IntegrityError, the FK field is nulled and retried.

    Args:
        records: Unified execution records to insert individually.
    """
    try:
        dropped = await self.hassette.database_service.submit(
            self.repository.persist_execution_batch_with_fk_fallback(records)
        )
        if dropped > 0:
            self._dropped_exhausted += dropped
            self.logger.error(
                "FK violation fallback: %d record(s) dropped even with null FK (total exhausted: %d)",
                dropped,
                self._dropped_exhausted,
            )
        else:
            await self.emit_completion_events(records)
    except Exception as exc:
        drop_count = len(records)
        self._dropped_exhausted += drop_count
        self.logger.error(
            "FK violation fallback failed entirely — dropping %d record(s) (total exhausted: %d): %s",
            drop_count,
            self._dropped_exhausted,
            exc,
        )