Skip to content

baldur.interfaces — Eventing, Notification & Audit

The unified event-bus protocol and its NoOp defaults, the notification and alert adapter contracts, and the audit-log adapter surface.

Event bus

EventBusProtocol

Bases: Protocol

Protocol for event bus implementations.

Both BaldurEventBus (L1 in-memory) and RedisEventBus (L2 distributed) implement this protocol. Used as the return type of the unified get_event_bus() factory.

Notification

NotificationAdapter

Bases: ABC

ABC for notification adapters.

Implement this class to send notifications to your preferred channel. Duck-typed adapters are also accepted via register_notification_adapter(), which auto-registers them as virtual subclasses.

Example

class SlackNotificationAdapter(NotificationAdapter): def send(self, payload: NotificationPayload) -> bool: response = requests.post( SLACK_WEBHOOK_URL, json={"text": f"{payload.title}\n{payload.message}"} ) return response.ok

def send_batch(self, payloads: list[NotificationPayload]) -> int:
    return sum(1 for p in payloads if self.send(p))

@property
def channel(self) -> NotificationChannel:
    return NotificationChannel.SLACK

channel abstractmethod property 

channel: NotificationChannel

Return the channel this adapter handles.

send abstractmethod 

send(payload: NotificationPayload) -> bool

Send a single notification.

Returns:

Type Description
bool

True if sent successfully, False otherwise

send_batch abstractmethod 

send_batch(payloads: list[NotificationPayload]) -> int

Send multiple notifications.

Returns:

Type Description
int

Number of successfully sent notifications

is_available 

is_available() -> bool

Check if this adapter is available (configuration validity).

Non-abstract concrete method — ABC-inheriting adapters inherit the default return True. Network-based adapters should override with I/O-free self-diagnosis (e.g., URL format validation).

Duck-typed adapters (virtual subclasses) do NOT inherit this — callers must use getattr(adapter, 'is_available', lambda: True)().

NotificationChannel module-attribute 

NotificationChannel = MessageChannel

NotificationSeverity module-attribute 

NotificationSeverity = MessageSeverity

Alerting

AlertSeverity module-attribute 

AlertSeverity = MessageSeverity

AlertCategory

Bases: str, Enum

Alert categories for routing.

Alert dataclass 

Alert(
    title: str,
    description: str,
    severity: AlertSeverity = AlertSeverity.WARNING,
    category: AlertCategory = AlertCategory.AVAILABILITY,
    timestamp: datetime = (lambda: utc_now())(),
    source: str = "baldur",
    service_name: str | None = None,
    domain: str | None = None,
    slo_name: str | None = None,
    slo_target: float | None = None,
    slo_current: float | None = None,
    details: dict[str, Any] = dict(),
    runbook_url: str | None = None,
    alert_key: str | None = None,
)

Bases: SerializableMixin

Alert containing all relevant context.

Captures: - What happened (title, description) - How severe (severity) - What category (category) - Where (source, service_name) - Additional context (details)

key property 

key: str

Generate alert key for deduplication.

to_json 

to_json() -> str

Convert to JSON string.

AlertAdapter

Bases: ABC

Abstract interface for alerting.

Implementations can send alerts to: - stdout (StdoutAlertAdapter) - Files (FileAlertAdapter) - Slack/Teams (user implements) - PagerDuty/OpsGenie (user implements) - Email (user implements) - Nowhere (NullAlertAdapter)

send abstractmethod 

send(alert: Alert) -> None

Send an alert.

Parameters:

Name Type Description Default
alert Alert

The alert to send

 required

resolve abstractmethod 

resolve(alert_key: str) -> None

Resolve/close an alert.

Parameters:

Name Type Description Default
alert_key str

The key of the alert to resolve

 required

alert_cb_opened 

alert_cb_opened(
    service_name: str,
    failure_count: int,
    threshold: int,
    is_manual: bool = False,
) -> None

Convenience method for Circuit Breaker open alert.

alert_cb_closed 

alert_cb_closed(
    service_name: str, is_manual: bool = False
) -> None

Convenience method for Circuit Breaker close (resolves alert).

alert_dlq_threshold 

alert_dlq_threshold(
    domain: str, pending_count: int, threshold: int
) -> None

Convenience method for DLQ threshold alert.

alert_slo_violation 

alert_slo_violation(
    slo_name: str,
    target: float,
    current: float,
    service_name: str | None = None,
) -> None

Convenience method for SLO violation alert.

alert_high_error_rate 

alert_high_error_rate(
    service_name: str, error_rate: float, threshold: float
) -> None

Convenience method for high error rate alert.

alert_failsafe_activated 

alert_failsafe_activated(
    component: str,
    error_message: str,
    fallback_action: str = "PROCEED",
) -> None

CRITICAL: Fail-Safe mode activation alert.

Sent when part of the Baldur system fails and transitions into Fail-Safe mode. This alert requires immediate attention.

Parameters:

Name Type Description Default
component str

Component that failed (e.g., "error_budget", "circuit_breaker")

 required
error_message str

Failure cause message

 required
fallback_action str

Fallback action taken (e.g., "PROCEED", "ALLOW")

 'PROCEED'
Note

This alert is designed to prevent "silent failures". When Fail-Safe activates, the system continues operating, but the operations team must be notified immediately to address the root cause.

resolve_failsafe 

resolve_failsafe(component: str) -> None

Resolve the alert when Fail-Safe recovers.

alert_failsafe_recovered 

alert_failsafe_recovered(
    component: str,
    downtime_seconds: float,
    recovery_reason: str = "System recovered automatically",
) -> None

Recovery notification: sent when normal operation is restored from Fail-Safe mode.

Similar to PagerDuty/OpsGenie "resolved" events, actively notifies that the failure has been cleared.

Parameters:

Name Type Description Default
component str

Component that recovered (e.g., "error_budget", "circuit_breaker")

 required
downtime_seconds float

Failure duration (seconds)

 required
recovery_reason str

Description of the recovery cause

 'System recovered automatically'
Note

This alert prevents "silent recoveries". When a failure is cleared it is explicitly announced, so the operations team does not need to keep tracking the failure state.

alert_override_escalation 

alert_override_escalation(
    override_type: str,
    requester: str,
    reason: str,
    service_name: str | None = None,
    escalation_channel: str = "#governance",
    escalation_mention: str = "@cto @security",
) -> None

Override escalation alert: sent when deployment is overridden under Error Budget exhaustion.

Netflix CAB (Change Advisory Board) style — when a deployment is forced through while the Error Budget is exhausted, escalates to a senior decision-maker / governance channel.

Parameters:

Name Type Description Default
override_type str

Override type (hotfix, security_patch, business_critical, etc.)

 required
requester str

Override requester

 required
reason str

Override reason

 required
service_name str | None

Target service name

 None
escalation_channel str

Escalation channel (e.g., #governance)

 '#governance'
escalation_mention str

Mention target (e.g., @cto @security)

 '@cto @security'
Note

This alert provides visibility into "risky actions". Every action that bypasses the Error Budget policy must be tracked.

Audit

AuditAction

Bases: str, Enum

Standard audit action types.

DLQ_FORCE_REDRIVE class-attribute instance-attribute 

DLQ_FORCE_REDRIVE = 'dlq_force_redrive'

Operator cap-override re-drive of an at-cap (REQUIRES_REVIEW) entry — a privileged, audited action distinct from a normal replay.

API_ERROR class-attribute instance-attribute 

API_ERROR = 'api_error'

Error raised while handling an API request.

VALIDATION_FAILED class-attribute instance-attribute 

VALIDATION_FAILED = 'validation_failed'

Input validation failed.

AUTHORIZATION_DENIED class-attribute instance-attribute 

AUTHORIZATION_DENIED = 'authorization_denied'

Authorization denied (no permission).

FORENSIC_CAPTURE_COMPLETED class-attribute instance-attribute 

FORENSIC_CAPTURE_COMPLETED = 'forensic_capture_completed'

Forensic context recorded after a captured exception.

AuditEntry dataclass 

AuditEntry(
    action: AuditAction | str,
    timestamp: datetime = (lambda: utc_now())(),
    actor_id: str | None = None,
    actor_type: str = "system",
    actor_roles: list[str] = list(),
    context_type: ContextType = ContextType.UNKNOWN,
    target_type: str | None = None,
    target_id: str | None = None,
    service_name: str | None = None,
    domain: str | None = None,
    reason: str | None = None,
    details: dict[str, Any] = dict(),
    success: bool = True,
    error_message: str | None = None,
)

Audit log entry containing all relevant context.

Captures: - What happened (action) - Who did it (actor_id, actor_type) - sourced automatically from ActorContext - What was affected (target_type, target_id) - Why (reason) - Additional context (details)

Note

When actor_id and actor_type are not set explicitly, they are pulled from ActorContext automatically. This auto-tracks "who set this and when".

to_dict 

to_dict() -> dict[str, Any]

Convert to dictionary for serialization.

to_json 

to_json() -> str

Convert to JSON string.

from_dict classmethod 

from_dict(data: dict[str, Any]) -> AuditEntry

Inverse of to_dict() — round-trip safe.

Unknown fields (e.g. integrity, checksum, audit_id added by ResilientContinuousAuditRecorder) are preserved in details under their original keys to maintain forensic completeness.

Forward-compatibility design (intentional, no schema_version field needed):

  1. New first-class field added later: old from_dict() running on new data — the unknown key falls into details, no exception. New from_dict() running on old data — data.get(key) returns None, default applies. Round-trip is preserved at every point in time.
  2. Field promoted from details to first-class: requires a one-time migration script. A schema_version field cannot solve this automatically.
  3. Field semantic change: cannot be auto-detected without a version field, but also cannot be auto-handled. Always requires coordinated reader/writer updates.

Conclusion: a schema_version meta field would add ceremony without solving the cases that actually need solving. The known-set + details-overflow design covers the only case that benefits from automation, and does so without ceremony.

Parameters:

Name Type Description Default
data dict[str, Any]

Dict produced by to_dict() (or a superset).

 required

Returns:

Type Description
AuditEntry

AuditEntry reconstructed from the dict.

from_wal_dict classmethod 

from_wal_dict(data: dict[str, Any]) -> AuditEntry

Convert a Write-Ahead-Log entry payload into an AuditEntry.

The audit WAL drained by the background sync worker (the recovery-replay path) holds the native WAL schema written by the log_*_audit helpers, not the to_dict() schema. That native shape carries event_type (not action) and a float-epoch timestamp (not an ISO string), so a plain from_dict() would drop the action and reset the timestamp to now — corrupting the two most audit-critical fields for a compliance trail. This converter maps the native schema faithfully and delegates the to_dict() shape to from_dict() as defense-in-depth for an injected or future-wired to_dict()-shape WAL.

The native-shape conversion is total: it never raises for a dict input — missing or malformed fields (a non-ISO/out-of-range timestamp, a non-mapping details) fall back to defaults — so a corrupted WAL entry can never become a poison entry that stalls the sync cursor.

Parameters:

Name Type Description Default
data dict[str, Any]

A WAL entry payload — the native log_*_audit shape (has event_type) or a to_dict() superset (has action).

 required

Returns:

Type Description
AuditEntry

AuditEntry reconstructed from the WAL payload.

AuditLogAdapter

Bases: ABC

Abstract interface for audit logging.

Implementations can store audit logs in: - Files (FileAuditLogAdapter) - stdout (StdoutAuditLogAdapter) - Database (user implements) - External services like Loki, Datadog (user implements) - Nowhere (NullAuditLogAdapter)

log abstractmethod 

log(entry: AuditEntry) -> None

Log an audit entry.

Parameters:

Name Type Description Default
entry AuditEntry

The audit entry to log

 required

query abstractmethod 

query(
    action: AuditAction | str | None = None,
    target_type: str | None = None,
    target_id: str | None = None,
    start_time: datetime | None = None,
    end_time: datetime | None = None,
    limit: int = 100,
) -> list[AuditEntry]

Query audit logs (optional - may not be supported by all adapters).

Parameters:

Name Type Description Default
action AuditAction | str | None

Filter by action type

 None
target_type str | None

Filter by target type

 None
target_id str | None

Filter by target ID

 None
start_time datetime | None

Filter from this time

 None
end_time datetime | None

Filter until this time

 None
limit int

Maximum entries to return

 100

Returns:

Type Description
list[AuditEntry]

List of matching audit entries

log_cb_open 

log_cb_open(
    service_name: str,
    reason: str,
    actor_id: str | None = None,
    is_manual: bool = True,
) -> None

Convenience method for Circuit Breaker open.

log_cb_close 

log_cb_close(
    service_name: str,
    reason: str,
    actor_id: str | None = None,
    is_manual: bool = True,
    trigger_replay: bool = False,
) -> None

Convenience method for Circuit Breaker close.

log_dlq_store 

log_dlq_store(
    dlq_id: str,
    domain: str,
    failure_type: str,
    error_message: str | None = None,
) -> None

Convenience method for DLQ storage.

dlq_id is the opaque entry id issued by the repository; it carries no numeric meaning and must not be parsed.

log_dlq_replay 

log_dlq_replay(
    dlq_id: str,
    domain: str,
    success: bool,
    actor_id: str | None = None,
    error_message: str | None = None,
) -> None

Convenience method for DLQ replay.

dlq_id is the opaque entry id issued by the repository; it carries no numeric meaning and must not be parsed.

log_retry 

log_retry(
    domain: str,
    func_name: str,
    attempt: int,
    max_attempts: int,
    success: bool,
    error_message: str | None = None,
) -> None

Convenience method for retry attempts.

log_governance_blocked 

log_governance_blocked(
    block_reason: str,
    operation_name: str,
    details: dict[str, Any] | None = None,
    service_name: str | None = None,
    domain: str | None = None,
) -> None

Record an entry when automation was blocked by governance.

This method answers the "why didn't the operation run at this time?" question concretely.

Parameters:

Name Type Description Default
block_reason str

Block reason (kill_switch, emergency_mode, error_budget)

 required
operation_name str

Name of the blocked operation

 required
details dict[str, Any] | None

Additional details (emergency_level, budget_percent, etc.)

 None
service_name str | None

Related service name

 None
domain str | None

Domain (payment, point, etc.)

 None