Notifications

Table of Contents

1. Overview
2. Architecture
3. Notification Types
4. Providers
5. Configuration
6. Flow Integration
7. API Endpoints
8. Deployment

---

Overview

Hegemony provides a flexible notification system that supports multiple delivery channels. Notifications can be triggered by:

• Flow-level subscriptions: Automatic notifications when run/approval events occur
• Explicit flow steps: Using the flow.notify handler in flow definitions

All notification delivery is handled by workers only - the API is responsible for CRUD operations on destinations/subscriptions and triggering notification workflows, but does not send notifications directly.

A destination's name follows the platform's object name rules: trimmed, not blank, at most 128 characters, no word longer than 64, and no control characters.

Key Principles

---

Architecture

High-Level Flow

Dispatch Paths

There are two main paths for notification dispatch:

1. Flow Execution Path (Workflow-Driven)

When a flow runs and changes status (started, completed, failed, etc.), the workflow dispatches notifications directly via Temporal activities.

2. API-Triggered Path (User Actions)

When a user initiates an action (cancel, pause, resume, approval decision), the API triggers a lightweight notification workflow.

3. Test Notification Path

When a user tests a destination, the API sends the request through the same Temporal workflow used for production dispatch. This ensures real end-to-end validation of the provider configuration. The workflow runs with skip_recording=True so no run-timeline event is created.

Workflow Input Fields

NotificationDispatchWorkflow accepts a NotificationWorkflowInput dataclass with the following fields:

Field	Type	Default	Description
run_id	str	required	Run ID (empty string for test sends)
event	str	required	Notification event type
destinations	list[dict]	required	Destination info dicts
run_context	dict | None	None	Run context for message formatting
approval_context	dict | None	None	Approval context for formatting
custom_title	str | None	None	Override the formatted title
custom_message	str | None	None	Override the formatted body
skip_recording	bool	False	Skip recording event in run timeline
org_id	str	""	The run's organization; scopes worker-side variable and secret resolution while formatting
org_slug	str	""	Slug of that organization; {{ secret('orgs/<slug>/…') }} references are confined to it
shared_org_slug	str	""	Slug of the designated shared organization when its secret namespace is admitted for this dispatch: under shared_resources (the default) only when the run, its flow, or one of the destinations belongs to the shared org; always under HEGEMONY_SHARED_ORG_SECRET_SCOPE=always

---

Notification Types

Run Lifecycle Events

Event	Trigger	Description
run.started	Run begins execution	Workflow starts running
run.paused	User pauses run	Execution paused, waiting for resume
run.resumed	User resumes run	Execution continues
run.cancelled	User cancels run	Execution stopped by user
run.failed	Step failure	Run failed during execution
run.completed	All steps succeed	Run finished successfully

Approval Events

Event	Trigger	Description
approval.requested	Approval step reached	Waiting for human approval
approval.approved	User approves	Approval granted, run continues
approval.rejected	User rejects	Approval denied, run fails
approval.expired	Timeout reached	No decision within timeout

---

Providers

Email (SMTP)

Sends notifications via SMTP server. Supports:

• Multiple recipients (To/Cc)
• Custom subject prefixes
• Plain text messages
• TLS/STARTTLS encryption

Configuration:

Field names use smtp_*_ref. Values are plain text, or secret references using Jinja template syntax ({{ secret('vault://path/key') }}) for anything sensitive. env() and file() are refused: they read the platform's own environment (see Secrets).

# Destination config_json
{
  "to": ["[email protected]", "[email protected]"],
  "cc": ["[email protected]"],
  "from": "[email protected]",
  "smtp_host_ref": "smtp.example.com",
  "smtp_port_ref": "587",
  "smtp_username_ref": "{{ secret('vault://orgs/default/secrets/smtp/username') }}",
  "smtp_password_ref": "{{ secret('vault://orgs/default/secrets/smtp/password') }}",
  "smtp_use_tls_ref": "false",
  "smtp_use_starttls_ref": "true"
}

Required fields and defaults:

to, from and smtp_host_ref are required; a destination without a sender address is refused when it sends. An empty port, TLS or STARTTLS field takes a fixed default: port 587, TLS off, STARTTLS on. There are no platform-wide SMTP settings: each destination carries its own.

Shoutrrr

Universal notification provider supporting many services via URL-based configuration. The URL is stored as a secret reference ({{ secret('vault://path/key') }}) for security.

Supported Services:

• Discord
• Slack
• Microsoft Teams
• Telegram
• Pushover
• Gotify
• Generic webhooks
• And many more

Configuration:

# Destination config_json
{
  "url_secret": "{{ secret('vault://orgs/default/secrets/discord/url') }}",  # Secret reference for the URL
  "title": "Hegemony Alert"           # Optional override
}

url_secret can also be a templated Shoutrrr URL when a provider stores credentials in separate secret keys. For example, Telegram bot token and chat ID values can be resolved independently:

url_secret: >-
  telegram://{{ secret('vault://orgs/default/secrets/jt-telegram-bot/token') }}@telegram?chats={{ secret('vault://orgs/default/secrets/jt-telegram-bot/groupid') }}

Outgoing Webhook

Sends HTTP POST/PUT requests to arbitrary URLs with configurable authentication, TLS, retry, and payload options. Supports signing policies for HMAC-authenticated endpoints.

Configuration:

# Destination config_json
{
  "url": "https://hooks.example.com/inbound/abc123",
  "method": "POST",                          # POST or PUT (default: POST)
  "headers": {"X-Source": "hegemony"},        # Extra request headers (optional)
  "payload_template": null,                   # Custom body; null = default JSON
  "timeout_seconds": 30,                      # Request timeout (default: 30)
  "verify_tls": true,                         # TLS verification (default: true)
  "custom_ca_bundle": null,                   # PEM CA text (optional, max 64 KB)
  "retry_count": 3,                           # Number of retries (default: 3)
  "retry_backoff_seconds": 2,                 # Backoff base (default: 2)
  "auth_type": "bearer",                      # none | bearer | basic | custom_header
  "auth_token_ref": "{{ secret('vault://orgs/default/secrets/my-token/value') }}",
  "signing": null                              # Signing policy (see below)
}

Authentication modes:

auth_type	Required fields	Description
none	—	No auth header
bearer	auth_token_ref	Authorization: Bearer <token>
basic	auth_username, auth_password_ref	Authorization: Basic <b64>
custom_header	auth_header_name, auth_header_value_ref	Arbitrary header

All *_ref fields are secret references ({{ secret('vault://...') }}).

Payload template:

When payload_template is set, {{title}} and {{body}} placeholders are replaced before sending. When null, the default payload is:

{"title": "<title>", "body": "<body>"}

This makes it easy to integrate with services like Power Automate (Adaptive Cards), PagerDuty, or any webhook-based API.

Signing policy:

An optional signing object adds HMAC-SHA256 signatures to outgoing requests. This is used when the receiving endpoint requires signed payloads (e.g. another Hegemony instance, GitHub-style webhooks).

signing:
  algorithm: "hmac-sha256"                    # Only supported value
  secret_ref: "{{ secret('vault://orgs/default/secrets/hmac-key/value') }}"
  signature_header: "X-Webhook-Signature"     # Header for the signature
  timestamp_header: "X-Hegemony-Timestamp"    # Optional: adds Unix timestamp header
  signed_content: "body"                      # "body" or "timestamp_dot_body"
  prefix: "sha256="                           # Prepended to hex digest (optional)

signed_content	Message signed
body	Raw request body
timestamp_dot_body	<timestamp>.<body> (requires timestamp_header)

Common presets:

Preset	signature_header	timestamp_header	signed_content	prefix
Generic HMAC	X-Webhook-Signature	—	body	sha256=
Hegemony-compatible	X-Hegemony-Signature	X-Hegemony-Timestamp	timestamp_dot_body	—
GitHub-style	X-Hub-Signature-256	—	body	sha256=

---

Configuration

Creating a Destination

POST /api/notifications/destinations
Content-Type: application/json

{
  "name": "Ops Team Email",
  "type": "email",
  "enabled": true,
  "config_json": {
    "to": ["[email protected]"]
  }
}

Creating, editing, and deleting a destination is written to the audit log, and the entry's before and after snapshots include config_json, so a change of address, webhook URL, or signing policy is visible as a diff. Credential-shaped keys are redacted on the way in, so a password or token in the configuration is never stored in the entry.

Creating a Subscription

POST /api/flows/{flow_id}/notifications/subscriptions
Content-Type: application/json

{
  "destination_id": "uuid-of-destination",
  "event": "run.failed",
  "enabled": true
}

Flow Step Notification

nodes:
  - id: notify_success
    type: step
    name: "Notify Success"
    tags: {phase: CLEANUP, kind: ACTION}
    handler: flow.notify
    params:
      destination_id: "uuid-of-destination"
      title: "Upgrade Complete for {{ run.name }}"
      message: |
        Flow: {{ run.flow_name }}
        Run: {{ run.name }}
        {% if target_lines %}
        Targets:
        {% for line in target_lines %}{{ line }}
        {% endfor %}{% endif %}

Available template variables for flow.notify:

• {{ run.id }}, {{ run.name }}, {{ run.flow_name }}, {{ run.status }} - Run details
• {{ event }}, {{ event_display }} - Notification event (always notification.sent for this handler)
• {{ target_lines }} - Pre-formatted target device summary lines (iterate with {% for line in target_lines %})
• {{ steps['node_id'].summary }} - Outputs of preceding steps, keyed by node id
• {{ ui_base_url }} - Base URL for run links
• {{ vars.name }}, {{ secret('...') }} - Config variables and secret references

---

Flow Integration

SendNotificationHandler

The flow.notify handler (class SendNotificationHandler in the hegemony-steps-flow plugin wheel) allows explicit notifications from within flows.

Each execution emits a run timeline event with kind=notification and includes the destination name plus success/failure details.

Parameters:

• destination_id (required): UUID of the notification destination
• title (optional): Custom notification title (supports Jinja2 templates)
• message (optional): Custom message body (supports Jinja2 templates)

Template Variables:

For the flow.notify handler (on-demand step notifications), see the variable list under Flow Step Notification above.

For run lifecycle events (run.completed, run.failed, etc.):

• {{ run.name }} - Run name
• {{ run.status }} - Run status
• {{ run.flow_name }} - Flow name
• {{ run.error_message }} - Error message (if failed)
• {{ run.failed_at_step_id }} - Step where failure occurred
• {{ run.display_targets_json }} - Target devices summary

For approval events:

• {{ approval.title }} - Approval request title
• {{ approval.message }} - Approval message
• {{ approval.status }} - Approval status
• {{ approval.decided_by }} - Who decided the approval
• {{ approval.decision_comment }} - Decision comment

---

API Endpoints

Public Endpoints

Method	Path	Description
GET	/api/notifications/destinations	List all destinations
POST	/api/notifications/destinations	Create destination
GET	/api/notifications/destinations/{id}	Get destination
PATCH	/api/notifications/destinations/{id}	Update destination
DELETE	/api/notifications/destinations/{id}	Delete destination
POST	/api/notifications/destinations/test-config	Test notification ad-hoc
POST	/api/notifications/destinations/{id}/test	Test existing destination
GET	/api/flows/{id}/notifications/subscriptions	List flow subscriptions
POST	/api/flows/{id}/notifications/subscriptions	Create subscription
PATCH	/api/flows/{id}/notifications/subscriptions/{sub_id}	Update subscription
DELETE	/api/flows/{id}/notifications/subscriptions/{sub_id}	Delete subscription

Internal Endpoints (Worker Only)

Method	Path	Description
GET	/internal/notification-destinations/{id}	Get destination for dispatch
GET	/internal/flows/{id}/notification-subscriptions	Get flow subscriptions

---

Deployment

Worker Configuration

Worker containers must include:

1. Shoutrrr binary - Installed automatically via Dockerfile
2. Secret backends referenced by destinations - the configured dynamic backends that {{ secret() }} refs point at

Environment Variables Summary

# UI Base URL - For notification links
# Also available as pydantic setting: get_settings().ui_base_url
HEGEMONY_UI_BASE_URL=https://hegemony.example.com

# Shoutrrr timeout
# Fixed provider constant: 30 seconds (not env-configurable)

Docker Compose Example

services:
  worker:
    build:
      context: ../..
      dockerfile: deploy/compose/Dockerfile.worker
    environment:
      # Core settings
      - HEGEMONY_TEMPORAL_HOST=temporal:7233
      - HEGEMONY_API_BASE_URL=http://api:8000

Multi-Worker Safety

When running multiple worker instances:

1. Temporal scheduling - Activities are distributed across workers by Temporal's task queue
2. Retry semantics - Notification dispatch uses Temporal retry policies (at-least-once); retries may occur on transient failures
3. Deduplication - Unique workflow IDs prevent duplicate workflow starts for the same event; however, activity retries can cause duplicate sends if the provider does not support idempotency

Note: To prevent duplicate notifications, ensure your notification providers support idempotency or implement deduplication on the receiving end.

---

Troubleshooting

Common Issues

Problem	Solution
Email not sending	Verify required references (for example smtp_host_ref) are set and all configured smtp_*_ref references resolve correctly
Shoutrrr failing	Verify url_secret reference resolves and format is correct
Outgoing webhook timeout	Check timeout_seconds, TLS settings, and target reachability from the worker container
Outgoing webhook 401	Verify auth_token_ref / signing secret_ref resolve; check secret value matches the receiver
Notifications not appearing	Check flow has subscriptions for the event type
Missing run events	Ensure worker can reach API (HEGEMONY_API_BASE_URL)

Logs

Worker logs include notification dispatch details:

INFO: Dispatching notification: run_id=xxx, event=run.completed, destinations=2
INFO: Notification sent: event=run.completed, destination=Ops Email, type=email
INFO: Notification sent: event=run.completed, destination=Slack Alerts, type=shoutrrr
INFO: Outgoing webhook delivered: url=https://hooks.example.com/inbound/… status=200

Testing

Use the test endpoint to verify destination configuration:

POST /api/notifications/destinations/test-config
Content-Type: application/json

{
  "type": "email",
  "config_json": {
    "to": ["[email protected]"]
  }
}

---

Database Schema

The notifications feature uses the following tables (defined in apps/api/models.py):

• notification_destinations — stores configured destinations (type, config JSON, enabled flag)
• flow_notification_subscriptions — links flows to destinations with event filters