POMDPPlanners.simulations.simulations_deployment.run_progress package

Progress tracking + notification module for long experiment runs.

Two layers:

Layer 1 (in-process): a Notifier constructed automatically by BaseSimulator. Emits run_started / episode_completed (heartbeat) / run_finished / run_failed events to a local SQLite DB and (if a Slack webhook is configured) to Slack.
Layer 2 (external watcher): a CLI module (POMDPPlanners.simulations.simulations_deployment.run_progress.watcher) that reads the same DB and emits run_stalled Slack notifications for runs whose last heartbeat is older than a configurable threshold. Run from cron to catch hard process death (SIGKILL / OOM / reboot).

Whether notifications are sent is controlled by the deployment environment: build_notifier() inspects SLACK_WEBHOOK_URL and a few opt-out env vars. User-facing code constructs SimulationsAPI as usual; no configuration is threaded through the public API.

class POMDPPlanners.simulations.simulations_deployment.run_progress.NotificationConfig(webhook_url=None, trial_interval=0, progress_db_path=None, disable=False)[source]

Bases: object

Configuration for Slack + progress-DB notifications.

Construct directly for programmatic per-instance control, or via from_env() to mirror the legacy env-var-driven behaviour at the API layer.

Parameters:

webhook_url (str | None)
trial_interval (int)
progress_db_path (Path | None)
disable (bool)

webhook_url: Slack incoming-webhook URL. None means “no Slack posts” (events still write to the progress DB if a notifier is constructed, but the SlackNotifier is replaced with a NullNotifier).

trial_interval: Number of completed Optuna trials between milestone Slack messages during hyperparameter tuning. 0 disables milestone messages. Only the HyperParameterOptimizer reads this; the simulator ignores it.

progress_db_path: Optional override for the SQLite progress DB path. None falls back to the path resolved by ProgressDB (POMDP_PROGRESS_DB env var, then ~/.cache/POMDPPlanners/progress.db).

disable: Hard kill switch. When True, the notifier built from this config is always a NullNotifier, even if webhook_url is set. Equivalent to the legacy POMDPPLANNERS_DISABLE_NOTIFY=1 env var.

Example

Enabling Slack notifications has two equivalent paths.

Zero-config (env var). Export SLACK_WEBHOOK_URL and any SimulationsAPI constructed without an explicit notification_config will pick it up automatically via from_env():

# In the shell:
#   export SLACK_WEBHOOK_URL="https://hooks.slack.com/services/..."
from POMDPPlanners.simulations.simulation_apis.local_simulations_api \
    import LocalSimulationsAPI
api = LocalSimulationsAPI()  # notifications enabled

Other env vars read by from_env(): POMDPPLANNERS_DISABLE_NOTIFY=1 (silence a single invocation even if a webhook is set), HYPERPARAM_TRIAL_NOTIFICATION_INTERVAL (Optuna milestone cadence; default 50, set 0 to disable), POMDP_PROGRESS_DB (override the SQLite progress DB path). Pytest runs are auto-silenced via PYTEST_CURRENT_TEST.

Programmatic (per-instance control). Construct explicitly to route multiple simulations in the same process to different Slack channels, or to override env-driven defaults:

from POMDPPlanners.simulations.simulations_deployment.run_progress \
    import NotificationConfig
cfg_a = NotificationConfig(webhook_url="https://hooks.slack.com/A")
cfg_b = NotificationConfig(webhook_url="https://hooks.slack.com/B")
api_a = LocalSimulationsAPI(notification_config=cfg_a)
api_b = LocalSimulationsAPI(notification_config=cfg_b)

Stall detection (external watcher). The in-process notifier cannot report a death the process itself does not witness (SIGKILL / OOM / host reboot). Run this once a minute from cron to catch those:

* * * * * /path/to/.venv/bin/python -m \
    POMDPPlanners.simulations.simulations_deployment.run_progress.watcher \
    --threshold-seconds 3600

--threshold-seconds is the heartbeat age above which a still- running entry is reported as stalled.

disable: bool = False

classmethod disabled()[source]

Return a config with notifications hard-disabled.

Used as the default for simulation classes constructed without an explicit config (e.g. unit tests, workflows called outside a SimulationsAPI).

Return type:: NotificationConfig
Returns:: A NotificationConfig with disable=True.

classmethod from_env()[source]

Build a config from the run-progress environment variables.

Reads SLACK_WEBHOOK_URL, POMDPPLANNERS_DISABLE_NOTIFY, PYTEST_CURRENT_TEST, HYPERPARAM_TRIAL_NOTIFICATION_INTERVAL, and POMDP_PROGRESS_DB. Returns a config that is functionally disabled (disable=True) when running under pytest or when POMDPPLANNERS_DISABLE_NOTIFY=1 is set; otherwise reflects the webhook URL (or None if unset) and the milestone interval (default 50).

Return type:: NotificationConfig
Returns:: A new NotificationConfig instance.

is_active()[source]

Return True iff this config should produce a real Slack notifier.

A config is active when it is not hard-disabled and has a non-empty webhook URL. When False, build_notifier() returns a NullNotifier.

Return type:: bool

progress_db_path: Path | None = None

trial_interval: int = 0

webhook_url: str | None = None

class POMDPPlanners.simulations.simulations_deployment.run_progress.Notifier(*args, **kwargs)[source]

Bases: Protocol

Protocol describing the four run-lifecycle events a simulator emits.

Implementations may write to a database, post to a chat tool, both, or neither. The Protocol is runtime_checkable() so tests can assert structural conformance without inheritance.

episode_completed(run_id)[source]

Record that a single episode finished (heartbeat).

Parameters:: run_id (str) – The run to which the episode belongs.
Return type:: None

run_failed(run_id, error)[source]

Record that a run aborted with an error.

Parameters:

run_id (str) – The run that failed.
error (str) – Stringified exception or signal-name describing the cause.

Return type:

None

run_finished(run_id)[source]

Record that a run finished cleanly.

Parameters:: run_id (str) – The run to mark as finished.
Return type:: None

run_started(run_id, *, metadata=None)[source]

Record that a run has begun.

Parameters:

run_id (str) – Unique id for the run.
metadata (dict[str, Any] | None) – Optional metadata dict describing the run.

Return type:

None

class POMDPPlanners.simulations.simulations_deployment.run_progress.NullNotifier[source]

Bases: object

No-op Notifier. All methods discard their arguments.

Used when notifications are disabled (SLACK_WEBHOOK_URL unset, POMDPPLANNERS_DISABLE_NOTIFY=1 set, or running under pytest).

Example

Basic usage:

notifier = NullNotifier()
notifier.run_started("rid", metadata={"k": "v"})
notifier.episode_completed("rid")
notifier.run_finished("rid")

episode_completed(run_id)[source]

No-op.

Return type:: None
Parameters:: run_id (str)

run_failed(run_id, error)[source]

No-op.

Return type:

None

Parameters:

run_id (str)
error (str)

run_finished(run_id)[source]

No-op.

Return type:: None
Parameters:: run_id (str)

run_started(run_id, *, metadata=None)[source]

No-op.

Return type:

None

Parameters:

run_id (str)
metadata (dict[str, Any] | None)

class POMDPPlanners.simulations.simulations_deployment.run_progress.ProgressDB(path=None)[source]

Bases: object

SQLite-backed progress writer / reader for experiment runs.

Opens a fresh connection per call (no shared connection state), so each method is safe to call from any thread inside the parent process. WAL journaling is enabled on first connect so that a separate watcher process can read concurrently. The path can be supplied explicitly or resolved from the POMDP_PROGRESS_DB environment variable; if neither is set, the default is ~/.cache/POMDPPlanners/progress.db.

Example

Basic usage:

db = ProgressDB()
db.start_run("abc123", "tiger_experiment", {"episodes": 100})
db.heartbeat("abc123")
db.finish_run("abc123", status="finished", error=None)

Parameters:: path (Path | None)

finish_run(run_id, status, error=None)[source]

Mark a run finished or failed.

Parameters:

run_id (str) – The run to update.
status (Literal['running', 'finished', 'failed']) – Either "finished" or "failed".
error (str | None) – Optional error message; stored verbatim.

Return type:

None

get_run(run_id)[source]

Fetch a single run by id.

Parameters:: run_id (str) – The run to look up.
Return type:: RunRow | None
Returns:: The matching RunRow, or None if no such run exists.

heartbeat(run_id)[source]

Bump the heartbeat timestamp for a run.

Parameters:: run_id (str) – The run to update.
Return type:: None

list_stalled(threshold_seconds)[source]

Return runs whose last heartbeat is older than threshold_seconds.

Only status='running' rows that have not yet been stall-notified are returned, so the watcher can use this list directly without additional filtering.

Parameters:: threshold_seconds (int) – Heartbeat-age threshold in seconds. A run is considered stalled if now - last_heartbeat_at exceeds this.
Return type:: list[RunRow]
Returns:: List of RunRow entries, oldest heartbeat first.

mark_stall_notified(run_id)[source]

Record that a stall notification has been emitted for a run.

Parameters:: run_id (str) – The run to mark.
Return type:: None

start_run(run_id, experiment_name, metadata=None)[source]

Record the start of a new run.

Parameters:

run_id (str) – Unique id for the run.
experiment_name (str) – Human-readable label for the run.
metadata (dict[str, Any] | None) – Optional metadata dict, JSON-serialized into storage.

Return type:

None

class POMDPPlanners.simulations.simulations_deployment.run_progress.RunRow(run_id, experiment_name, host, pid, status, started_at, finished_at, last_heartbeat_at, error_msg, metadata, stall_notified_at)[source]

Bases: object

A snapshot of one row in the runs table.

Parameters:

run_id (str)
experiment_name (str)
host (str)
pid (int)
status (str)
started_at (str)
finished_at (str | None)
last_heartbeat_at (str)
error_msg (str | None)
metadata (dict[str, Any])
stall_notified_at (str | None)

run_id: Unique id for the run (typically uuid4().hex).

experiment_name: Human-readable label for the run (e.g. the simulator experiment name).

host: Host the run started on, as returned by socket.gethostname().

pid: Process id of the parent simulator process.

status: One of "running", "finished", or "failed".

started_at: ISO-8601 UTC timestamp of run start.

finished_at: ISO-8601 UTC timestamp of run finish, or None while running.

last_heartbeat_at: ISO-8601 UTC timestamp of the most recent heartbeat.

error_msg: Stringified exception if the run failed, else None.

metadata: JSON-decoded metadata dict supplied at run start.

stall_notified_at: ISO-8601 UTC timestamp at which a watcher sent a stall notification for this run, or None if no stall notification has been sent.

error_msg: str | None

experiment_name: str

finished_at: str | None

host: str

last_heartbeat_at: str

metadata: dict[str, Any]

pid: int

run_id: str

stall_notified_at: str | None

started_at: str

status: str

class POMDPPlanners.simulations.simulations_deployment.run_progress.SlackNotifier(webhook_url, experiment_name, *, db=None, logger=None)[source]

Bases: object

Notifier that writes to ProgressDB and Slack.

run_started / run_finished / run_failed write to the DB and POST a short message to the Slack webhook. episode_completed only writes a heartbeat to the DB — it does not post to Slack, to avoid spamming the channel during long runs.

Slack-side failures are caught and logged at WARNING level via the supplied logger; they do not propagate to the caller.

Example

Basic usage:

notifier = SlackNotifier(
    webhook_url="https://hooks.slack.com/services/T/B/X",
    experiment_name="tiger_experiment",
)
notifier.run_started("rid-1", metadata={"episodes": 100})
notifier.episode_completed("rid-1")
notifier.run_finished("rid-1")

Parameters:

webhook_url (str)
experiment_name (str)
db (ProgressDB | None)
logger (logging.Logger | None)

episode_completed(run_id)[source]

Update the heartbeat in the DB. Does not post to Slack.

Parameters:: run_id (str) – The run to which the episode belongs.
Return type:: None

run_failed(run_id, error)[source]

Mark the run failed in the DB and POST a failure message.

Parameters:

run_id (str) – The run that failed.
error (str) – Stringified exception or signal-name describing the cause.

Return type:

None

run_finished(run_id)[source]

Mark the run finished in the DB and POST a success message.

Parameters:: run_id (str) – The run to mark as finished.
Return type:: None

run_started(run_id, *, metadata=None)[source]

Record start in DB and POST a start message to Slack.

Parameters:

run_id (str) – Unique id for the run.
metadata (dict[str, Any] | None) – Optional metadata dict written to the DB.

Return type:

None

POMDPPlanners.simulations.simulations_deployment.run_progress.build_notifier(experiment_name, *, config, logger=None)[source]

Construct the appropriate Notifier from a config.

This function performs no environment reads. The caller passes an explicit NotificationConfig; if that config is active (non-empty webhook URL AND disable=False), a SlackNotifier is returned. Otherwise a NullNotifier is returned. Env-var resolution belongs in NotificationConfig.from_env() — typically invoked by the SimulationsAPI layer.

Parameters:

experiment_name (str) – Human-readable label used for run records and Slack messages.
config (NotificationConfig) – The NotificationConfig controlling whether and where Slack messages are sent.
logger (Logger | None) – Optional logger forwarded to SlackNotifier.

Return type:

Notifier

Returns:

A Notifier instance — either NullNotifier or SlackNotifier depending on whether config.is_active().

POMDPPlanners.simulations.simulations_deployment.run_progress.install_signal_handlers(notifier, run_id)[source]

Install SIGTERM/SIGINT handlers that emit Notifier.run_failed().

The installed handlers:

Call notifier.run_failed(run_id, error="signal:<NAME>"). Exceptions inside the notifier are swallowed.
Chain to the previous handler if it is callable. Non-callable handlers (signal.SIG_DFL / signal.SIG_IGN, both integer sentinels) are skipped, and the signal is re-delivered to the process with default handling restored.

Returns a callable that uninstalls both handlers (restoring whatever was previously installed). Calling the uninstall function multiple times is safe.

Parameters:

notifier (Notifier) – The Notifier to emit run_failed on signal.
run_id (str) – The run identifier to pass to run_failed.

Return type:

Callable[[], None]

Returns:

An idempotent uninstall callable. The caller should invoke it during normal teardown (e.g. BaseSimulator.__exit__) so signal handling reverts to the prior configuration.

POMDPPlanners.simulations.simulations_deployment.run_progress.post_trial_milestone(webhook_url, *, experiment_name, run_id, completed_trials, total_trials, config_name, best_score, logger=None)[source]

POST a Slack message reporting hyperparameter-tuning progress.

Stateless helper called once per HYPERPARAM_TRIAL_NOTIFICATION_INTERVAL Optuna trials. webhook_url must be a non-empty Slack incoming-webhook URL — callers (typically the in-task Optuna callback) are responsible for skipping the call when notifications are disabled. Network errors are caught and logged at WARNING level; nothing propagates to the caller.

Parameters:

webhook_url (str) – Slack incoming-webhook URL.
experiment_name (str) – Human-readable label for the tuning run, included in the message body.
run_id (str) – Unique id of the parent tuning run (same id used by run_started/run_finished).
completed_trials (int) – Number of trials completed globally so far.
total_trials (int) – Total trials scheduled across all configs.
config_name (str) – Name of the config whose trial just completed; lets the reader see which leg of the sweep is currently running.
best_score (float | None) – Best objective value seen so far in the current config (e.g. study.best_value). Pass None if no trial has produced a usable score yet — the message will note this.
logger (Logger | None) – Optional logger for swallowed-error WARNING. Defaults to a logger named run_progress.notify.

Return type:

None

Submodules

POMDPPlanners.simulations.simulations_deployment.run_progress.config module

Notification configuration dataclass for the run_progress module.

NotificationConfig bundles every parameter the run-progress infrastructure needs into a single explicit object. Simulation classes (BaseSimulator, HyperParameterOptimizer) accept one as a constructor argument; the SimulationsAPI layer is the only thing that reads env vars (via NotificationConfig.from_env()).

This split keeps simulation classes pure DI — they no longer touch os.environ at construction time — while preserving the existing zero-config UX through the API layer.

class POMDPPlanners.simulations.simulations_deployment.run_progress.config.NotificationConfig(webhook_url=None, trial_interval=0, progress_db_path=None, disable=False)[source]

Bases: object

Configuration for Slack + progress-DB notifications.

Construct directly for programmatic per-instance control, or via from_env() to mirror the legacy env-var-driven behaviour at the API layer.

Parameters:

webhook_url (str | None)
trial_interval (int)
progress_db_path (Path | None)
disable (bool)

webhook_url: Slack incoming-webhook URL. None means “no Slack posts” (events still write to the progress DB if a notifier is constructed, but the SlackNotifier is replaced with a NullNotifier).

trial_interval: Number of completed Optuna trials between milestone Slack messages during hyperparameter tuning. 0 disables milestone messages. Only the HyperParameterOptimizer reads this; the simulator ignores it.

progress_db_path: Optional override for the SQLite progress DB path. None falls back to the path resolved by ProgressDB (POMDP_PROGRESS_DB env var, then ~/.cache/POMDPPlanners/progress.db).

disable: Hard kill switch. When True, the notifier built from this config is always a NullNotifier, even if webhook_url is set. Equivalent to the legacy POMDPPLANNERS_DISABLE_NOTIFY=1 env var.

Example

Enabling Slack notifications has two equivalent paths.

Zero-config (env var). Export SLACK_WEBHOOK_URL and any SimulationsAPI constructed without an explicit notification_config will pick it up automatically via from_env():

# In the shell:
#   export SLACK_WEBHOOK_URL="https://hooks.slack.com/services/..."
from POMDPPlanners.simulations.simulation_apis.local_simulations_api \
    import LocalSimulationsAPI
api = LocalSimulationsAPI()  # notifications enabled

Other env vars read by from_env(): POMDPPLANNERS_DISABLE_NOTIFY=1 (silence a single invocation even if a webhook is set), HYPERPARAM_TRIAL_NOTIFICATION_INTERVAL (Optuna milestone cadence; default 50, set 0 to disable), POMDP_PROGRESS_DB (override the SQLite progress DB path). Pytest runs are auto-silenced via PYTEST_CURRENT_TEST.

Programmatic (per-instance control). Construct explicitly to route multiple simulations in the same process to different Slack channels, or to override env-driven defaults:

from POMDPPlanners.simulations.simulations_deployment.run_progress \
    import NotificationConfig
cfg_a = NotificationConfig(webhook_url="https://hooks.slack.com/A")
cfg_b = NotificationConfig(webhook_url="https://hooks.slack.com/B")
api_a = LocalSimulationsAPI(notification_config=cfg_a)
api_b = LocalSimulationsAPI(notification_config=cfg_b)

Stall detection (external watcher). The in-process notifier cannot report a death the process itself does not witness (SIGKILL / OOM / host reboot). Run this once a minute from cron to catch those:

* * * * * /path/to/.venv/bin/python -m \
    POMDPPlanners.simulations.simulations_deployment.run_progress.watcher \
    --threshold-seconds 3600

--threshold-seconds is the heartbeat age above which a still- running entry is reported as stalled.

disable: bool = False

classmethod disabled()[source]

Return a config with notifications hard-disabled.

Used as the default for simulation classes constructed without an explicit config (e.g. unit tests, workflows called outside a SimulationsAPI).

Return type:: NotificationConfig
Returns:: A NotificationConfig with disable=True.

classmethod from_env()[source]

Build a config from the run-progress environment variables.

Reads SLACK_WEBHOOK_URL, POMDPPLANNERS_DISABLE_NOTIFY, PYTEST_CURRENT_TEST, HYPERPARAM_TRIAL_NOTIFICATION_INTERVAL, and POMDP_PROGRESS_DB. Returns a config that is functionally disabled (disable=True) when running under pytest or when POMDPPLANNERS_DISABLE_NOTIFY=1 is set; otherwise reflects the webhook URL (or None if unset) and the milestone interval (default 50).

Return type:: NotificationConfig
Returns:: A new NotificationConfig instance.

is_active()[source]

Return True iff this config should produce a real Slack notifier.

A config is active when it is not hard-disabled and has a non-empty webhook URL. When False, build_notifier() returns a NullNotifier.

Return type:: bool

progress_db_path: Path | None = None

trial_interval: int = 0

webhook_url: str | None = None

POMDPPlanners.simulations.simulations_deployment.run_progress.db module

SQLite-backed progress storage for long experiment runs.

Stores per-run metadata, status transitions, and a heartbeat timestamp that gets bumped once per completed episode. The DB is written to only by the parent process of a run (joblib workers fork, and SQLite connections do not survive fork). Reads from a separate watcher process are safe thanks to WAL mode plus a 5-second busy timeout.

Classes:: ProgressDB: SQLite progress writer / reader. RunRow: Frozen dataclass mirroring one row of the runs table.

class POMDPPlanners.simulations.simulations_deployment.run_progress.db.ProgressDB(path=None)[source]

Bases: object

SQLite-backed progress writer / reader for experiment runs.

Opens a fresh connection per call (no shared connection state), so each method is safe to call from any thread inside the parent process. WAL journaling is enabled on first connect so that a separate watcher process can read concurrently. The path can be supplied explicitly or resolved from the POMDP_PROGRESS_DB environment variable; if neither is set, the default is ~/.cache/POMDPPlanners/progress.db.

Example

Basic usage:

db = ProgressDB()
db.start_run("abc123", "tiger_experiment", {"episodes": 100})
db.heartbeat("abc123")
db.finish_run("abc123", status="finished", error=None)

Parameters:: path (Path | None)

finish_run(run_id, status, error=None)[source]

Mark a run finished or failed.

Parameters:

run_id (str) – The run to update.
status (Literal['running', 'finished', 'failed']) – Either "finished" or "failed".
error (str | None) – Optional error message; stored verbatim.

Return type:

None

get_run(run_id)[source]

Fetch a single run by id.

Parameters:: run_id (str) – The run to look up.
Return type:: RunRow | None
Returns:: The matching RunRow, or None if no such run exists.

heartbeat(run_id)[source]

Bump the heartbeat timestamp for a run.

Parameters:: run_id (str) – The run to update.
Return type:: None

list_stalled(threshold_seconds)[source]

Return runs whose last heartbeat is older than threshold_seconds.

Only status='running' rows that have not yet been stall-notified are returned, so the watcher can use this list directly without additional filtering.

Parameters:: threshold_seconds (int) – Heartbeat-age threshold in seconds. A run is considered stalled if now - last_heartbeat_at exceeds this.
Return type:: list[RunRow]
Returns:: List of RunRow entries, oldest heartbeat first.

mark_stall_notified(run_id)[source]

Record that a stall notification has been emitted for a run.

Parameters:: run_id (str) – The run to mark.
Return type:: None

start_run(run_id, experiment_name, metadata=None)[source]

Record the start of a new run.

Parameters:

run_id (str) – Unique id for the run.
experiment_name (str) – Human-readable label for the run.
metadata (dict[str, Any] | None) – Optional metadata dict, JSON-serialized into storage.

Return type:

None

class POMDPPlanners.simulations.simulations_deployment.run_progress.db.RunRow(run_id, experiment_name, host, pid, status, started_at, finished_at, last_heartbeat_at, error_msg, metadata, stall_notified_at)[source]

Bases: object

A snapshot of one row in the runs table.

Parameters:

run_id (str)
experiment_name (str)
host (str)
pid (int)
status (str)
started_at (str)
finished_at (str | None)
last_heartbeat_at (str)
error_msg (str | None)
metadata (dict[str, Any])
stall_notified_at (str | None)

run_id: Unique id for the run (typically uuid4().hex).

experiment_name: Human-readable label for the run (e.g. the simulator experiment name).

host: Host the run started on, as returned by socket.gethostname().

pid: Process id of the parent simulator process.

status: One of "running", "finished", or "failed".

started_at: ISO-8601 UTC timestamp of run start.

finished_at: ISO-8601 UTC timestamp of run finish, or None while running.

last_heartbeat_at: ISO-8601 UTC timestamp of the most recent heartbeat.

error_msg: Stringified exception if the run failed, else None.

metadata: JSON-decoded metadata dict supplied at run start.

stall_notified_at: ISO-8601 UTC timestamp at which a watcher sent a stall notification for this run, or None if no stall notification has been sent.

error_msg: str | None

experiment_name: str

finished_at: str | None

host: str

last_heartbeat_at: str

metadata: dict[str, Any]

pid: int

run_id: str

stall_notified_at: str | None

started_at: str

status: str

POMDPPlanners.simulations.simulations_deployment.run_progress.notify module

Notifier interface, implementations, and signal-handler installer.

The Notifier Protocol defines four events that the simulator emits during a run:

run_started — fired at the entry point of the experiment loop.
episode_completed — fired once per completed episode (heartbeat).
run_finished — fired on a clean exit.
run_failed — fired when an exception or signal aborts the run.

Two concrete implementations are provided:

NullNotifier — no-op; used when notifications are disabled.
SlackNotifier — writes events to the local progress DB and posts to a Slack incoming webhook (via stdlib urllib.request).

The build_notifier() factory consumes a NotificationConfig and returns the appropriate implementation; it does not read any environment variables itself — env-var resolution is the responsibility of the caller (typically NotificationConfig.from_env() invoked by the SimulationsAPI layer). The install_signal_handlers() helper attaches SIGTERM/SIGINT handlers that emit run_failed before re-raising.

class POMDPPlanners.simulations.simulations_deployment.run_progress.notify.Notifier(*args, **kwargs)[source]

Bases: Protocol

Protocol describing the four run-lifecycle events a simulator emits.

Implementations may write to a database, post to a chat tool, both, or neither. The Protocol is runtime_checkable() so tests can assert structural conformance without inheritance.

episode_completed(run_id)[source]

Record that a single episode finished (heartbeat).

Parameters:: run_id (str) – The run to which the episode belongs.
Return type:: None

run_failed(run_id, error)[source]

Record that a run aborted with an error.

Parameters:

run_id (str) – The run that failed.
error (str) – Stringified exception or signal-name describing the cause.

Return type:

None

run_finished(run_id)[source]

Record that a run finished cleanly.

Parameters:: run_id (str) – The run to mark as finished.
Return type:: None

run_started(run_id, *, metadata=None)[source]

Record that a run has begun.

Parameters:

run_id (str) – Unique id for the run.
metadata (dict[str, Any] | None) – Optional metadata dict describing the run.

Return type:

None

class POMDPPlanners.simulations.simulations_deployment.run_progress.notify.NullNotifier[source]

Bases: object

No-op Notifier. All methods discard their arguments.

Used when notifications are disabled (SLACK_WEBHOOK_URL unset, POMDPPLANNERS_DISABLE_NOTIFY=1 set, or running under pytest).

Example

Basic usage:

notifier = NullNotifier()
notifier.run_started("rid", metadata={"k": "v"})
notifier.episode_completed("rid")
notifier.run_finished("rid")

episode_completed(run_id)[source]

No-op.

Return type:: None
Parameters:: run_id (str)

run_failed(run_id, error)[source]

No-op.

Return type:

None

Parameters:

run_id (str)
error (str)

run_finished(run_id)[source]

No-op.

Return type:: None
Parameters:: run_id (str)

run_started(run_id, *, metadata=None)[source]

No-op.

Return type:

None

Parameters:

run_id (str)
metadata (dict[str, Any] | None)

class POMDPPlanners.simulations.simulations_deployment.run_progress.notify.SlackNotifier(webhook_url, experiment_name, *, db=None, logger=None)[source]

Bases: object

Notifier that writes to ProgressDB and Slack.

run_started / run_finished / run_failed write to the DB and POST a short message to the Slack webhook. episode_completed only writes a heartbeat to the DB — it does not post to Slack, to avoid spamming the channel during long runs.

Slack-side failures are caught and logged at WARNING level via the supplied logger; they do not propagate to the caller.

Example

Basic usage:

notifier = SlackNotifier(
    webhook_url="https://hooks.slack.com/services/T/B/X",
    experiment_name="tiger_experiment",
)
notifier.run_started("rid-1", metadata={"episodes": 100})
notifier.episode_completed("rid-1")
notifier.run_finished("rid-1")

Parameters:

webhook_url (str)
experiment_name (str)
db (ProgressDB | None)
logger (logging.Logger | None)

episode_completed(run_id)[source]

Update the heartbeat in the DB. Does not post to Slack.

Parameters:: run_id (str) – The run to which the episode belongs.
Return type:: None

run_failed(run_id, error)[source]

Mark the run failed in the DB and POST a failure message.

Parameters:

run_id (str) – The run that failed.
error (str) – Stringified exception or signal-name describing the cause.

Return type:

None

run_finished(run_id)[source]

Mark the run finished in the DB and POST a success message.

Parameters:: run_id (str) – The run to mark as finished.
Return type:: None

run_started(run_id, *, metadata=None)[source]

Record start in DB and POST a start message to Slack.

Parameters:

run_id (str) – Unique id for the run.
metadata (dict[str, Any] | None) – Optional metadata dict written to the DB.

Return type:

None

POMDPPlanners.simulations.simulations_deployment.run_progress.notify.build_notifier(experiment_name, *, config, logger=None)[source]

Construct the appropriate Notifier from a config.

This function performs no environment reads. The caller passes an explicit NotificationConfig; if that config is active (non-empty webhook URL AND disable=False), a SlackNotifier is returned. Otherwise a NullNotifier is returned. Env-var resolution belongs in NotificationConfig.from_env() — typically invoked by the SimulationsAPI layer.

Parameters:

experiment_name (str) – Human-readable label used for run records and Slack messages.
config (NotificationConfig) – The NotificationConfig controlling whether and where Slack messages are sent.
logger (Logger | None) – Optional logger forwarded to SlackNotifier.

Return type:

Notifier

Returns:

A Notifier instance — either NullNotifier or SlackNotifier depending on whether config.is_active().

POMDPPlanners.simulations.simulations_deployment.run_progress.notify.install_signal_handlers(notifier, run_id)[source]

Install SIGTERM/SIGINT handlers that emit Notifier.run_failed().

The installed handlers:

Call notifier.run_failed(run_id, error="signal:<NAME>"). Exceptions inside the notifier are swallowed.
Chain to the previous handler if it is callable. Non-callable handlers (signal.SIG_DFL / signal.SIG_IGN, both integer sentinels) are skipped, and the signal is re-delivered to the process with default handling restored.

Returns a callable that uninstalls both handlers (restoring whatever was previously installed). Calling the uninstall function multiple times is safe.

Parameters:

notifier (Notifier) – The Notifier to emit run_failed on signal.
run_id (str) – The run identifier to pass to run_failed.

Return type:

Callable[[], None]

Returns:

An idempotent uninstall callable. The caller should invoke it during normal teardown (e.g. BaseSimulator.__exit__) so signal handling reverts to the prior configuration.

POMDPPlanners.simulations.simulations_deployment.run_progress.notify.post_trial_milestone(webhook_url, *, experiment_name, run_id, completed_trials, total_trials, config_name, best_score, logger=None)[source]

POST a Slack message reporting hyperparameter-tuning progress.

Stateless helper called once per HYPERPARAM_TRIAL_NOTIFICATION_INTERVAL Optuna trials. webhook_url must be a non-empty Slack incoming-webhook URL — callers (typically the in-task Optuna callback) are responsible for skipping the call when notifications are disabled. Network errors are caught and logged at WARNING level; nothing propagates to the caller.

Parameters:

webhook_url (str) – Slack incoming-webhook URL.
experiment_name (str) – Human-readable label for the tuning run, included in the message body.
run_id (str) – Unique id of the parent tuning run (same id used by run_started/run_finished).
completed_trials (int) – Number of trials completed globally so far.
total_trials (int) – Total trials scheduled across all configs.
config_name (str) – Name of the config whose trial just completed; lets the reader see which leg of the sweep is currently running.
best_score (float | None) – Best objective value seen so far in the current config (e.g. study.best_value). Pass None if no trial has produced a usable score yet — the message will note this.
logger (Logger | None) – Optional logger for swallowed-error WARNING. Defaults to a logger named run_progress.notify.

Return type:

None

POMDPPlanners.simulations.simulations_deployment.run_progress.watcher module

CLI watcher that posts Slack alerts for stalled experiment runs.

Intended to be invoked from cron (typically once per minute):

* * * * * /path/to/.venv/bin/python -m         POMDPPlanners.simulations.simulations_deployment.run_progress.watcher         --threshold-seconds 3600

For every run in the progress DB whose status is running AND whose last_heartbeat_at is older than --threshold-seconds AND which has not previously been stall-notified, the watcher POSTs a short Slack message and marks the row notified so subsequent ticks do not duplicate.

The watcher is stateless — it makes one pass through the DB and exits. Cron handles “the watcher died” semantics so no separate daemon is needed.

POMDPPlanners.simulations.simulations_deployment.run_progress.watcher.main(argv=None)[source]

Run one watcher pass.

Parameters:: argv (Optional[Sequence[str]]) – Optional argv override (for tests). When None, falls back to sys.argv after the program name.
Return type:: int
Returns:: Exit status. 0 on success (including the “no stalled runs” case). Non-zero is reserved for hard configuration errors.