Internal event bus to decouple `Task` status from logging,

[FernandoCelmer]

notifications, and metrics

Labels: enhancement

Context

dotflow/core/task.py:251-267 runs three side effects inside the status setter:

@status.setter
def status(self, value):
    self._status = value
    self.config.notify.hook_status_task(task=self)
    if value == FAILED:
        self.config.log.error(task=self)
        self.config.metrics.task_failed(task=self)
    elif value == RETRY:
        ...

Problems:

• A property assignment performs HTTP (notify), I/O (log), and a
  counter update (metrics).
• Tests that flip status need three mocks every time.
• Adding a new observer (Sentry tracer, audit log, custom webhook)
  means editing Task itself.
• Failure ordering is undocumented: notify runs first; if it raises,
  log and metric never execute.

Concept

Introduce a tiny in-process event bus. The setter mutates state and emits a StatusChanged event. Three default subscribers (LogSubscriber, NotifySubscriber, MetricsSubscriber) each handle the event independently and are registered in Config.

Proposed implementation

# dotflow/core/events.py
from collections.abc import Callable
from dataclasses import dataclass
from typing import Any
from dotflow.logging import logger

@dataclass
class StatusChanged:
    task: "Task"
    old: "TypeStatus"
    new: "TypeStatus"

class EventBus:
    def __init__(self) -> None:
        self._subs: list[Callable[[Any], None]] = []

    def subscribe(self, handler: Callable[[Any], None]) -> None:
        self._subs.append(handler)

    def emit(self, event: Any) -> None:
        for handler in self._subs:
            try:
                handler(event)
            except Exception:
                logger.exception("event handler failed")

# dotflow/core/task.py
@status.setter
def status(self, value):
    old = self._status
    self._status = value
    self.config.events.emit(StatusChanged(task=self, old=old, new=value))

# dotflow/core/config.py
from dotflow.core.events import EventBus
from dotflow.core.subscribers import (
    LogSubscriber, NotifySubscriber, MetricsSubscriber,
)

class Config:
    def __init__(self, ...):
        ...
        self.events = EventBus()
        self.events.subscribe(LogSubscriber(self.log))
        self.events.subscribe(NotifySubscriber(self.notify))
        self.events.subscribe(MetricsSubscriber(self.metrics))

A subscriber example:

class LogSubscriber:
    def __init__(self, log): self.log = log

    def __call__(self, event):
        if not isinstance(event, StatusChanged):
            return
        if event.new == TypeStatus.FAILED:
            self.log.error(task=event.task)
        elif event.new == TypeStatus.RETRY:
            self.log.warning(task=event.task)
        else:
            self.log.info(task=event.task)

Benefits

• Failure in one subscriber no longer cascades.
• Adding a new observer = events.subscribe(my_handler). No edits
  to Task.
• The bus naturally supports new event types
  (CheckpointSaved, WorkflowStarted) without re-touching Task.

Acceptance criteria

• EventBus, StatusChanged defined in
  dotflow/core/events.py
• LogSubscriber, NotifySubscriber, MetricsSubscriber
  defined in dotflow/core/subscribers.py
• Task.status setter emits the event and does no other I/O
• Config wires the three default subscribers
• Existing observable behavior preserved (verified by current
  tests)
• New test: a raising subscriber does not stop other subscribers

Out of scope

Async subscribers (handler runs in a worker thread). The synchronous bus is enough for the current observability surface.

---