Skip to content

baldur_pro.services.emergency_mode — Emergency Mode

System-wide emergency levels and the manager that activates them: EmergencyLevel, get_emergency_manager, and is_emergency_active.

🔒 PRO Feature — requires a baldur-pro license

These symbols ship in the baldur-pro distribution. PRO modules import normally — there is no ImportError. PRO features activate only when baldur.init() runs with a valid BALDUR_LICENSE_KEY; without it the system runs with OSS defaults and register_pro_services() logs entitlement.pro_registration_skipped.

emergency_mode

Emergency Mode Service — Single-instance emergency lifecycle.

Manages emergency mode activation/deactivation, traffic tier rules (Load Shedding), and gradual recovery for a single service instance.

For region-scoped namespace emergency management (regional isolation, cascade detection, partition reconciliation), see regional_emergency/.

Architecture

models.emergency.EmergencyLevel (shared domain type) │ ├── emergency_mode/ ← this package (single-instance lifecycle) └── regional_emergency/ (regional extension, depends on this)

Features: - EmergencyLevel: emergency levels (NORMAL, LEVEL_1, LEVEL_2, LEVEL_3) - GracefulDegradationManager: emergency mode activation/deactivation - RecoveryGate: metric-based recovery stabilization and gradual recovery

Usage

from baldur_pro.services.emergency_mode import ( get_emergency_manager, EmergencyLevel, is_emergency_active, )

Check emergency status

if is_emergency_active(): level = get_emergency_manager().get_current_level() ...

Activate emergency mode

get_emergency_manager().activate_manual( level=EmergencyLevel.LEVEL_2, reason="High error rate detected", activated_by="admin", duration_minutes=30, )

Deactivate emergency mode

get_emergency_manager().deactivate(deactivated_by="admin")

Status: Public

RecoveryGateConfig dataclass 

RecoveryGateConfig(
    stabilization_period_seconds: int = 300,
    require_metrics_stable: bool = True,
    cpu_threshold_percent: float = 80.0,
    error_rate_threshold: float = 0.05,
    gradual_recovery: bool = True,
    level_step_delay_seconds: int = 60,
    health_check_interval_seconds: int = 30,
    auto_rollback_on_failure: bool = True,
)

Bases: SerializableMixin

Stabilization-window configuration for safe emergency exit.

Defines how long the system must remain stable before the recovery gate releases emergency mode, plus the metric thresholds that count as "stable". Defaults are read from EmergencyModeSettings when available.

stabilization_period_seconds class-attribute instance-attribute 

stabilization_period_seconds: int = 300

Stabilization wait window (seconds).

require_metrics_stable class-attribute instance-attribute 

require_metrics_stable: bool = True

Whether metric-based stability checks are required.

cpu_threshold_percent class-attribute instance-attribute 

cpu_threshold_percent: float = 80.0

CPU usage must be at or below this to count as stable.

error_rate_threshold class-attribute instance-attribute 

error_rate_threshold: float = 0.05

Error rate must be at or below this (5%) to count as stable.

gradual_recovery class-attribute instance-attribute 

gradual_recovery: bool = True

Whether to step down emergency level gradually.

level_step_delay_seconds class-attribute instance-attribute 

level_step_delay_seconds: int = 60

Delay between level-down steps (seconds).

health_check_interval_seconds class-attribute instance-attribute 

health_check_interval_seconds: int = 30

Metric re-check cadence during recovery (seconds).

auto_rollback_on_failure class-attribute instance-attribute 

auto_rollback_on_failure: bool = True

Whether a failed mid-walk metrics check stops the gradual recovery and holds the current level. The name is historical: no level is restored.

from_settings classmethod 

from_settings() -> RecoveryGateConfig

Create from EmergencyModeSettings, with hardcoded fallback.

EmergencyLevel

Bases: str, Enum

Emergency level definitions.

Each level determines per-tier traffic multipliers. Ordering: NORMAL < LEVEL_1 < LEVEL_2 < LEVEL_3 (severity-based).

severity property 

severity: int

Numeric severity for ordering comparisons and backward compatibility.

from_severity classmethod 

from_severity(severity: int) -> EmergencyLevel

Create EmergencyLevel from numeric severity (0-3).

Supports legacy integer-based code that used IntEnum values.

EmergencyModeError 

EmergencyModeError(message: str = '', *, code: str = '')

Bases: BaldurError

Base exception for emergency mode operations.

EmergencyStateError 

EmergencyStateError(
    message: str, *, operation: str = "", detail: str = ""
)

Bases: EmergencyModeError

Invalid state or input for emergency mode operation.

Covers: missing parameters, inactive emergency mode, duplicate recovery, invalid target level.

RecoveryNotAllowedError 

RecoveryNotAllowedError(
    message: str,
    *,
    check_reason: str = "",
    open_breakers: tuple[str, ...] = ()
)

Bases: EmergencyModeError

Recovery gate policy prevents deactivation.

check_reason is the gate's own sentence; open_breakers names the breakers whose open or half-open state the gate counted, so an operator can act per breaker instead of forcing the release.

GracefulDegradationManager

Bases: EventEmitterMixin

Emergency-mode manager - stepwise entry into / exit from emergency mode.

Implemented as a thread-safe singleton.

Usage

manager = GracefulDegradationManager()

Manually activate emergency mode

manager.activate_manual( level=EmergencyLevel.LEVEL_2, reason="High error rate", activated_by="admin", duration_minutes=30, )

Check the current state

state = manager.get_state()

Deactivate emergency mode

manager.deactivate(deactivated_by="admin")

close 

close() -> None

Drop this manager's refresher registration. Idempotent.

get_state 

get_state() -> EmergencyState

Get the current state (this process's copy; no store I/O).

get_current_level 

get_current_level() -> EmergencyLevel

Get the current emergency-mode level (this process's copy).

is_active 

is_active() -> bool

Whether emergency mode is active (this process's copy).

get_refresh_status 

get_refresh_status() -> dict[str, Any]

This process's read health of the level, for the status response.

get_previous_states 

get_previous_states() -> list[dict[str, Any]]

Get the list of previous-state snapshots.

Returns:

Type Description
list[dict[str, Any]]

Snapshot list (newest first)

rollback_to_previous 

rollback_to_previous(
    index: int = 0,
) -> EmergencyState | None

Roll back to a previous state.

Restores the snapshot's level and activation fields on the stored state and clears the gradual-recovery fields: a rollback ends any walk and never revives one.

Parameters:

Name Type Description Default
index int

Snapshot index to roll back to (0=most recent, 1=the one before...)

 0

Returns:

Type Description
EmergencyState | None

The rolled-back state, or None when no such snapshot exists

Raises:

Type Description
SystemControlStoreError

The store did not confirm the rollback.

get_tier_multiplier 

get_tier_multiplier(tier_id: str) -> float

Return the tier multiplier for the current emergency-mode level.

Parameters:

Name Type Description Default
tier_id str

Tier ID (critical, standard, non_essential)

 required

Returns:

Type Description
float

Multiplier (0.0 ~ 1.0)

activate_manual 

activate_manual(
    level: EmergencyLevel,
    reason: str,
    activated_by: str,
    duration_minutes: int | None = None,
    is_chaos_experiment: bool = False,
    experiment_id: str | None = None,
    override_kill_switch: bool = False,
) -> EmergencyState

Manually activate emergency mode.

An activation ends a running gradual recovery (in every process — the walk's next step finds its identity gone).

Parameters:

Name Type Description Default
level EmergencyLevel

Emergency-mode level

 required
reason str

Activation reason (required)

 required
activated_by str

User who activated it

 required
duration_minutes int | None

Auto-expiry time (minutes); None requires manual deactivation

 None
is_chaos_experiment bool

Whether activation is from a chaos experiment

 False
experiment_id str | None

Related chaos experiment ID

 None
override_kill_switch bool

True to allow activation when kill switch is ON

 False

Returns:

Type Description
EmergencyState

The new state

Raises:

Type Description
SystemControlStoreError

The store did not confirm the activation; it is in force nowhere.

activate_auto 

activate_auto(
    level: EmergencyLevel,
    reason: str,
    duration_minutes: int | None = None,
) -> EmergencyState

Automatically activate emergency mode (system-triggered).

Automated activations respect the kill switch unconditionally, and never lower or repeat a level: the stored state — not this process's copy — decides, on every write attempt. A declined activation adopts the stored state it read, so this process acts on that level at once. A change the store did not confirm is logged and in force nowhere; the unchanged state is returned.

Parameters:

Name Type Description Default
level EmergencyLevel

Emergency-mode level

 required
reason str

Auto-detection reason

 required
duration_minutes int | None

Auto-expiry time (default 30 minutes)

 None

Returns:

Type Description
EmergencyState

The new state

deactivate 

deactivate(
    deactivated_by: str,
    reason: str = "",
    force: bool = False,
) -> EmergencyState

Deactivate emergency mode.

Parameters:

Name Type Description Default
deactivated_by str

User who deactivated it

 required
reason str

Deactivation reason (optional)

 ''
force bool

Force deactivation, ignoring recovery conditions

 False

Returns:

Type Description
EmergencyState

The new state

Raises:

Type Description
RecoveryNotAllowedError

The recovery gate refused.
SystemControlStoreError

The store did not confirm the release.

start_gradual_recovery 

start_gradual_recovery(
    initiated_by: str,
    target_level: EmergencyLevel = EmergencyLevel.NORMAL,
) -> EmergencyState

Start gradual recovery.

Eases stepwise from the current level down to the target level. Proceeds after a metrics check at each step. The walk carries an identity (recovery_id); a step lands only while the stored walk is still this one.

Parameters:

Name Type Description Default
initiated_by str

User who started the recovery

 required
target_level EmergencyLevel

Target level (default: NORMAL)

 NORMAL

Returns:

Type Description
EmergencyState

The current state

Raises:

Type Description
EmergencyStateError

Not active, a walk in progress, or a target not below the current level.
SystemControlStoreError

The store did not confirm the start.

stop_gradual_recovery 

stop_gradual_recovery(
    stopped_by: str, reason: str = ""
) -> EmergencyState

Stop gradual recovery — the operator's stop.

Clears the stored recovery fields whenever a walk is stored, whichever process runs it: the walking process ends at its next step, whose identity check then fails. Also clears a walk no thread runs any more. This process's walk thread is stopped only once the store holds the stop (or holds no walk): a stop the store did not confirm is in force nowhere, so the walk keeps running and the stored walk stays owned.

Raises:

Type Description
SystemControlStoreError

The store did not confirm the stop.

stop_gradual_recovery_on_shutdown 

stop_gradual_recovery_on_shutdown(
    stopped_by: str, reason: str = ""
) -> None

Stop this process's walk at shutdown; never raises.

Clears the stored recovery fields only when this process's walk thread is alive and the stored walk is its walk. A process that merely inherited a walk (a worker forked from the process running it) writes nothing.

get_history 

get_history(limit: int = 50) -> list[dict[str, Any]]

Get the change history (changes this process committed).

set_recovery_gate_config 

set_recovery_gate_config(
    config: RecoveryGateConfig, changed_by: str = "system"
)

Update the recovery gate configuration.

get_recovery_gate_config 

get_recovery_gate_config() -> RecoveryGateConfig

Get the current recovery gate configuration.

reset 

reset()

Reset state (for tests). A blind write of the defaults.

EmergencyState dataclass 

EmergencyState(
    level: EmergencyLevel = EmergencyLevel.NORMAL,
    is_active: bool = False,
    activated_at: str | None = None,
    activated_by: str | None = None,
    activation_reason: str | None = None,
    expires_at: str | None = None,
    deactivated_at: str | None = None,
    deactivated_by: str | None = None,
    is_auto_triggered: bool = False,
    is_recovering: bool = False,
    recovery_started_at: str | None = None,
    target_level: EmergencyLevel | None = None,
    recovery_id: str | None = None,
    metadata: dict[str, Any] | None = None,
)

Bases: SerializableMixin

Emergency-mode state.

metadata class-attribute instance-attribute 

metadata: dict[str, Any] | None = None

Additional metadata.

When activated by a chaos experiment

{ "is_chaos_experiment": True, "experiment_id": "exp-xxx", "classification": "chaos_induced_test" }

RecoveryCheck dataclass 

RecoveryCheck(
    allowed: bool,
    reason: str,
    failed_on: (
        Literal[
            "cpu", "error_rate", "unmeasured", "checker"
        ]
        | None
    ),
    tripped_names: tuple[str, ...] = (),
    excluded_tripped: bool = False,
)

One recovery-gate verdict with the structure behind its reason.

Attributes:

Name Type Description
allowed bool

Whether recovery may proceed.
reason str

The operator-facing sentence; carried verbatim by RecoveryNotAllowedError.check_reason and the REST 409.
failed_on Literal['cpu', 'error_rate', 'unmeasured', 'checker'] | None

Which check refused — "cpu", "error_rate", an "unmeasured" rate, or the "checker" raising — or None on a pass.
tripped_names tuple[str, ...]

The names whose breakers were open or half-open in the reading, whether they were counted or excluded.
excluded_tripped bool

True when the rate compared was measured over the names those breakers do not hold.

RecoveryGate 

RecoveryGate(
    config: RecoveryGateConfig | None = None,
    metrics_checker: (
        Callable[[], dict[str, Any]] | None
    ) = None,
)

Recovery gate - manages safe deactivation of emergency mode.

Features: - Metric-based stability checks - Gradual recovery (stepwise easing per level) - Automatic rollback on recovery failure

Parameters:

Name Type Description Default
config RecoveryGateConfig | None

Recovery gate configuration

 None
metrics_checker Callable[[], dict[str, Any]] | None

Callback returning the current system metrics Return value: {"cpu_percent": 75.0, "error_rate": 0.02} An error_rate of None (or a missing key) means "not measured" and refuses recovery; the optional error_rate_calls is the call count the rate was measured over, rendered in the pass reason when present. The optional error_rate_excluding_tripped / error_rate_calls_excluding_tripped pair and tripped_names let evaluate compare the rate over the names no open breaker holds.

 None

check_recovery_allowed 

check_recovery_allowed() -> tuple[bool, str]

Check whether recovery is allowed.

Fail-closed on every unreadable input: a checker that raises, and an error rate the checker could not measure (None or absent), both refuse — an absent measurement is never read as a healthy 0.0.

Returns:

Type Description
(is_allowed, reason)

whether recovery is allowed and the reason

evaluate 

evaluate(*, exclude_tripped: bool = False) -> RecoveryCheck

Run the recovery check and return its structured verdict.

The checks run in a fixed order — CPU, then the error rate — and the first breach refuses. Fail-closed on every unreadable input: a checker that raises, and an error rate the checker could not measure (None or absent), both refuse.

Parameters:

Name Type Description Default
exclude_tripped bool

Compare the error rate over the names no open breaker holds — the reading for the one emergency step that lets held breakers probe again, where those names' refusals cannot fall until the step is taken. Honoured only when the checker supplies the excluding pair; an injected checker without it is compared on the full rate.

 False

Returns:

Type Description
RecoveryCheck

The verdict, its reason and the names it counted or left out.

get_next_recovery_level 

get_next_recovery_level(
    current_level: EmergencyLevel,
) -> EmergencyLevel | None

Return the next level in gradual recovery.

LEVEL_3 -> LEVEL_2 -> LEVEL_1 -> NORMAL

Returns:

Type Description
EmergencyLevel | None

The next level, or None (when already NORMAL)

get_emergency_manager 

get_emergency_manager() -> GracefulDegradationManager

Get the emergency-mode manager singleton.

reset_emergency_manager 

reset_emergency_manager() -> None

Reset the emergency manager (clears both module-level and class-level singletons, joins recovery thread).

register_emergency_control_state 

register_emergency_control_state() -> None

Register the emergency level with this process's control-state refresher.

Called by the PRO startup extension, before baldur.init() loads the control state, so that load — and every worker's own — reads the level too. The manager is resolved lazily on the first pass; its construction performs no store I/O and re-registers the key with its own callback.

is_emergency_active 

is_emergency_active() -> bool

Check whether emergency mode is active (convenience function).

Usage

if is_emergency_active(): # Handle emergency mode ...

get_emergency_level 

get_emergency_level() -> EmergencyLevel

Get the current emergency-mode level (convenience function).

Usage

level = get_emergency_level() if level >= EmergencyLevel.LEVEL_2: # Handle a severe emergency ...

get_tier_multiplier 

get_tier_multiplier(tier_id: str) -> float

Get the tier multiplier for the current emergency mode (convenience function).

Usage

multiplier = get_tier_multiplier("standard") if random.random() > multiplier: # Load shedding return Response(status=503)