AI Agent Escalation: Page a Human When Limits Hit

Decide which conditions need a person, have the runtime check them on every run, and send a notification with enough context to act on when one trips, while the agent pauses. In Promptise, the budget, health and mission checks each take an EscalationTarget: a webhook_url that receives a JSON POST, an event_type emitted on your event bus, or both. Budget and health escalations also suspend the process until someone resumes it:

from promptise.runtime import BudgetConfig, EscalationTarget, ProcessConfig

config = ProcessConfig(
    instructions="You answer billing tickets.",
    budget=BudgetConfig(
        enabled=True, max_runs_per_day=200, on_exceeded="escalate",
        escalation=EscalationTarget(webhook_url="https://relay.example.com/escalate"),
    ),
)

Key takeaways:

• Three things escalate: a budget limit (budget_violation), a health anomaly (health_anomaly) and an unsure mission judge (low_confidence).

• Escalation pauses the work. Budget and health suspend the process; a mission escalation pauses the mission. Nothing resumes by itself.

• The webhook must be public. An SSRF guard refuses private and loopback addresses. For internal targets, use event_type and relay it yourself.

• The payload is fixed JSON. Slack and PagerDuty need their own formats, so put a small relay in between.

• Escalations are not notifications. One page-a-human message per incident, next to the event stream that reports everything.

Escalation is the agent handing a decision to a person because a rule says it shouldn't decide alone. In customer support it usually means a handoff: the bot passes the conversation to a human agent. For an autonomous agent that runs on triggers, it means something more operational: stop doing work, tell the on-call person why, and wait.

A good escalation has three parts. A trigger condition that is checked by code, not by the agent's own judgment, so a confused agent can't talk itself out of it. A payload that says which agent, what limit or anomaly, and the numbers, so the person can decide without digging through logs. And a safe state while waiting: no more tool calls, events kept in a queue.

This is how the pieces connect in the Promptise runtime:

budget limit refused a call or run ─┐
health check found an anomaly ──────┼─▶ escalate(target, payload)
mission judge below the threshold ──┘        │
                                             ├─▶ webhook_url: one JSON POST to a public address
                                             └─▶ event_type: event_bus.emit(type, payload) ─▶ your relay
                                                                                   (Slack, PagerDuty, tickets)
then:  budget, health ─▶ process SUSPENDED (reason "budget" / "health"); events queue up
       mission        ─▶ mission PAUSED; the process keeps running, the mission is not judged
a person fixes the cause ─▶ resume(): POST .../resume  or  POST .../mission/resume

Escalate when continuing could cost money, cause damage or waste a day:

• A daily cost, run or tool-call limit was hit, or an irreversible action limit would be crossed.

• The agent repeats the same tool call, loops, or keeps answering with nothing.

• A long mission is not converging and a person should look at the approach.

A notification is enough when nobody needs to act right away: a budget warning at 80%, a finished run, a recovered health check. Send those through event notifications, and keep pages for the cases above. Paging for everything trains people to ignore the pager.

For a decision inside a run, such as approving one refund, an escalation is the wrong tool. Use human approval on that tool, or let the agent ask a question through its inbox.

• Python 3.10 or later and Promptise Foundry 1.5.0 or later (httpx and aiohttp come with it).

• No API key and no Slack or PagerDuty account. A scripted model server, fake_llm.py, plays the model and the judge, and receiver.py plays Slack and PagerDuty on your machine.

• About 20 minutes.

pip install "promptise>=1.5.0"

escalate() is the function all three checks call. Calling it directly shows how each kind of target behaves, including the two ways an escalation can silently go nowhere:

"""A minimal in-memory event bus: the runtime calls emit(); your code subscribes."""
from dataclasses import dataclass, field
from uuid import uuid4


@dataclass
class BusEvent:
    data: dict
    source: str = ""
    event_id: str = field(default_factory=lambda: uuid4().hex)


class InMemoryBus:
    def __init__(self) -> None:
        self._handlers: dict[str, list] = {}

    async def subscribe(self, event_type, handler):
        self._handlers.setdefault(event_type, []).append(handler)

    async def unsubscribe(self, event_type, handler):
        self._handlers.get(event_type, []).remove(handler)

    async def emit(self, event_type, payload):
        for handler in list(self._handlers.get(event_type, [])):
            await handler(BusEvent(data=payload))

"""Step 1: what escalate() does with each kind of target."""
import asyncio
import logging
import sys

from promptise.runtime import EscalationTarget
from promptise.runtime.escalation import escalate

from ops_bus import InMemoryBus

logging.basicConfig(stream=sys.stdout, level=logging.WARNING, format="  %(levelname)s %(name)s: %(message)s")
payload = {"type": "budget_violation", "process_name": "billing-bot"}


async def main() -> None:
    print("1. webhook on this machine:")
    await escalate(EscalationTarget(webhook_url="http://127.0.0.1:9785/escalate"), payload)

    print("2. event type, but no event bus:")
    await escalate(EscalationTarget(event_type="ops.escalation"), payload)

    print("3. event type on a bus:")
    bus = InMemoryBus()

    async def on_call(event):
        print("  received:", event.data)

    await bus.subscribe("ops.escalation", on_call)
    await escalate(EscalationTarget(event_type="ops.escalation"), payload, event_bus=bus)

    print("4. not a URL:")
    try:
        EscalationTarget(webhook_url="hooks.slack.com/services/T000")
    except ValueError as exc:
        print("  ValueError:", str(exc).split("Value error, ")[1].split(" [type")[0])


asyncio.run(main())

1. webhook on this machine:
  WARNING promptise.runtime.escalation: Escalation SSRF blocked: '127.0.0.1' resolves to private/internal IP 127.0.0.1
2. event type, but no event bus:
  WARNING promptise.runtime.escalation: Escalation event 'ops.escalation' not emitted: the process has no event bus
  WARNING promptise.runtime.escalation: Escalation (budget_violation) has no target: set EscalationTarget.webhook_url or event_type
3. event type on a bus:
  received: {'type': 'budget_violation', 'process_name': 'billing-bot'}
4. not a URL:
  ValueError: webhook_url must be an http:// or https:// URL, got 'hooks.slack.com/services/T000'

Four things to take from this:

1. A webhook on your own machine is refused. The host is resolved before every POST, and any private, loopback or link-local address is blocked, so a compromised config can't make the runtime call internal services or cloud metadata. Escalation webhooks have no opt-out. A local test receiver needs the event bus route.

2. `event_type` needs an event bus. Without one, the escalation is logged as a warning and dropped. Pass a bus to AgentProcess(event_bus=...) or AgentRuntime(event_bus=...).

3. The bus is yours. The runtime doesn't ship one. Any object with async subscribe, unsubscribe and emit works, so you can wrap Redis, NATS or Kafka the same way.

4. Escalation never raises. Failures are warnings on the promptise.runtime.escalation logger, so the agent keeps its safe state even when delivery fails. Alert on those log lines.

Now let the governance checks escalate on their own. Three processes share one bus target, ops.escalation. The billing agent may run twice a day and gets three tickets. The triage agent's model returns empty replies. The migrator's judge reports low confidence:

"""Steps 2-4: budget, health and mission escalations, delivered on an event bus."""
import asyncio
import json
import logging
import os

os.environ["OPENAI_BASE_URL"] = "http://127.0.0.1:8781/v1"
os.environ["OPENAI_API_KEY"] = "not-a-real-key"
logging.basicConfig(level=logging.CRITICAL)

from fake_llm import FakeLLM
from ops_bus import InMemoryBus
from promptise.runtime import (
    AgentProcess, BudgetConfig, EscalationTarget, HealthConfig, MissionConfig,
    ProcessConfig, TriggerEvent,
)

ON_CALL = EscalationTarget(event_type="ops.escalation")


def ticket(n: int) -> TriggerEvent:
    return TriggerEvent(trigger_id="helpdesk", trigger_type="webhook", payload={"ticket": n})


async def scenario(llm, bus, name, config, events, reply="Done."):
    llm.reply = reply
    process = AgentProcess(name, config, event_bus=bus)
    await process.start()
    for n in range(1, events + 1):
        await process.inject(ticket(n))
        await asyncio.sleep(0.3)
    print(f"   -> {name}: state={process.state.value} suspend_reason={process.suspend_reason} "
          f"runs={process.status()['invocation_count']}")
    await process.stop()


async def main() -> None:
    llm = FakeLLM()
    llm.judge = {"achieved": False, "confidence": 0.35,
                 "reasoning": "Only 2 of 14 tables migrated; errors in orders_v2."}
    await llm.start()
    bus = InMemoryBus()

    async def on_call(event):
        print("escalation:", json.dumps(event.data))

    await bus.subscribe("ops.escalation", on_call)

    print("budget: max 2 runs a day, 3 tickets")
    await scenario(llm, bus, "billing-bot", ProcessConfig(
        instructions="You answer billing tickets.",
        budget=BudgetConfig(enabled=True, max_runs_per_day=2,
                            on_exceeded="escalate", escalation=ON_CALL),
    ), events=3)

    print("health: two empty replies in a row")
    await scenario(llm, bus, "triage-bot", ProcessConfig(
        instructions="You triage tickets.",
        health=HealthConfig(enabled=True, empty_threshold=2,
                            on_anomaly="escalate", escalation=ON_CALL),
    ), events=2, reply="")

    print("mission: the judge is not confident")
    await scenario(llm, bus, "migrator", ProcessConfig(
        instructions="Migrate one table per run.",
        mission=MissionConfig(enabled=True, objective="Migrate all 14 tables to schema v2",
                              success_criteria="All tables pass v2 validation",
                              confidence_threshold=0.7, escalation=ON_CALL),
    ), events=1)
    await llm.stop()


asyncio.run(main())

budget: max 2 runs a day, 3 tickets
escalation: {"type": "budget_violation", "process_id": "57639997-e095-4822-85a9-a7a53dd79b96", "process_name": "billing-bot", "violation": {"limit_name": "max_runs_per_day", "limit_value": 2, "current_value": 3, "tool_name": null}}
   -> billing-bot: state=suspended suspend_reason=budget runs=2
health: two empty replies in a row
escalation: {"type": "health_anomaly", "process_id": "63fde649-f300-4dda-af6b-63d90fbb9ade", "process_name": "triage-bot", "anomaly_type": "empty_response", "details": {"consecutive_count": 2, "max_chars": 10, "lengths": [0, 0]}}
   -> triage-bot: state=suspended suspend_reason=health runs=2
mission: the judge is not confident
escalation: {"type": "low_confidence", "process_id": "f57514e8-6160-4fd3-beb7-980f6e25d3ec", "confidence": 0.35, "reasoning": "Only 2 of 14 tables migrated; errors in orders_v2."}
   -> migrator: state=running suspend_reason=None runs=1

Each check sent exactly one escalation:

• Budget. The third run was refused before it reached the model, the violation was escalated, and the process was suspended with suspend_reason="budget". tool_name is null because a run limit, not a tool call, was crossed. For a tool-call or cost limit it names the tool.

• Health. Two empty replies in a row (empty_threshold=2) were an empty_response anomaly. The details say why: two replies of length 0, against a limit of 10 characters. The process was suspended with suspend_reason="health". A cooldown, 300 seconds by default, stops the same anomaly from paging again on every run.

• Mission. The judge said "not achieved" with confidence 0.35, below the 0.7 threshold. The payload carries the judge's reasoning. The process stays running, but the mission is paused and not judged again until someone resumes it, so it escalates once per pause.

What each payload holds:

process_id is the process UUID. Because the mission payload has no name, a relay that serves several agents should keep a map from ID to name: {p.process_id: name for name, p in runtime.processes.items()}.

Neither service accepts the raw payload. A Slack incoming webhook needs a text field and answers no_text without it. The PagerDuty Events API v2 (https://events.pagerduty.com/v2/enqueue) needs a routing_key, an event_action and a payload with summary, source and severity. So the escalation goes to a relay that formats it. This one subscribes to the bus, builds both messages, adds a dedup_key so repeats update one incident, retries failed deliveries, and does its work in a background task so a slow endpoint never holds up the agent:

"""Turn escalation payloads into a Slack message and a PagerDuty Events API v2 event."""
import asyncio
import logging
import os

import httpx

logger = logging.getLogger("on_call_relay")

# Real values come from your secret store; the defaults point at the local receiver.
SLACK_URL = os.environ.get("SLACK_WEBHOOK_URL", "http://127.0.0.1:9785/slack")
PAGERDUTY_URL = os.environ.get("PAGERDUTY_EVENTS_URL", "http://127.0.0.1:9785/pagerduty")
ROUTING_KEY = os.environ.get("PAGERDUTY_ROUTING_KEY", "local-test-key")

SEVERITY = {"budget_violation": "error", "health_anomaly": "error", "low_confidence": "warning"}


def describe(p: dict) -> tuple[str, str]:
    """A one-line summary and a dedup key for an escalation payload."""
    who = p.get("process_name") or p.get("process_id", "unknown")
    if p["type"] == "budget_violation":
        v = p["violation"]
        what = f"{v['limit_name']} hit ({v['current_value']}/{v['limit_value']})"
        key = v["limit_name"]
    elif p["type"] == "health_anomaly":
        what, key = f"{p['anomaly_type']} anomaly", p["anomaly_type"]
    else:
        what, key = f"mission confidence {p['confidence']:.2f}: {p['reasoning']}", "mission"
    return f"{who}: {what}", f"{who}:{p['type']}:{key}"


def to_slack(p: dict) -> dict:
    summary, _ = describe(p)
    return {"text": f":rotating_light: Agent needs a human. {summary}"}


def to_pagerduty(p: dict) -> dict:
    summary, dedup_key = describe(p)
    return {
        "routing_key": ROUTING_KEY,
        "event_action": "trigger",
        "dedup_key": dedup_key,  # repeats update one incident instead of opening new ones
        "payload": {"summary": summary[:1024], "source": "promptise-runtime",
                    "severity": SEVERITY.get(p["type"], "warning"), "custom_details": p},
    }


class OnCallRelay:
    """An event-bus subscriber. Delivery runs in the background so a slow
    endpoint never holds up the process that escalated."""

    def __init__(self) -> None:
        self._tasks: set[asyncio.Task] = set()

    async def __call__(self, event) -> None:
        task = asyncio.create_task(self._deliver(event.data))
        self._tasks.add(task)
        task.add_done_callback(self._tasks.discard)

    async def _deliver(self, payload: dict) -> None:
        async with httpx.AsyncClient(timeout=10) as client:
            for url, body in ((SLACK_URL, to_slack(payload)), (PAGERDUTY_URL, to_pagerduty(payload))):
                for attempt in range(3):
                    try:
                        r = await client.post(url, json=body)
                        if r.status_code < 300:
                            break
                    except httpx.HTTPError:
                        pass
                    await asyncio.sleep(2 ** attempt)
                else:
                    logger.error("escalation not delivered to %s", url)

    async def drain(self) -> None:
        await asyncio.gather(*self._tasks)

The URLs and routing key come from the environment. Unset, they point at a local receiver, so the whole path runs without credentials. The script also attaches an event notifier with a WebhookSink for process.suspended, to show the two channels side by side, and then tries two ways of resuming:

"""Step 5: a budget escalation reaches Slack and PagerDuty (here: a local receiver)."""
import asyncio
import logging
import os

os.environ["OPENAI_BASE_URL"] = "http://127.0.0.1:8781/v1"
os.environ["OPENAI_API_KEY"] = "not-a-real-key"
logging.basicConfig(level=logging.CRITICAL)

from fake_llm import FakeLLM
from on_call_relay import OnCallRelay
from ops_bus import InMemoryBus
from promptise.events import EventNotifier, WebhookSink
from promptise.runtime import (
    AgentRuntime, BudgetConfig, EscalationTarget, ProcessConfig, TriggerEvent,
)
from receiver import Receiver


async def main() -> None:
    llm, receiver = FakeLLM(), Receiver()
    await llm.start()
    await receiver.start()

    bus, relay = InMemoryBus(), OnCallRelay()
    await bus.subscribe("ops.escalation", relay)
    # For comparison: the event notifier reports the suspension that follows.
    notifier = EventNotifier(sinks=[WebhookSink(
        "http://127.0.0.1:9785/notifier", events=["process.suspended"],
        allow_private_networks=True,  # a receiver on this machine
        transform=lambda e: {"event": e["event_type"], "process": e["agent_id"],
                             "reason": e["data"]["reason"]},
    )])
    runtime = AgentRuntime(event_bus=bus, event_notifier=notifier)
    process = await runtime.add_process("billing-bot", ProcessConfig(
        instructions="You answer billing tickets.",
        budget=BudgetConfig(enabled=True, max_runs_per_day=2, on_exceeded="escalate",
                            escalation=EscalationTarget(event_type="ops.escalation")),
    ))
    await runtime.start_all()
    async def ticket(n: int) -> None:
        await process.inject(TriggerEvent(trigger_id="helpdesk", trigger_type="webhook",
                                          payload={"ticket": n}))
        await asyncio.sleep(0.3)
        await relay.drain()

    for n in (1, 2, 3):
        await ticket(n)
    await asyncio.sleep(0.3)

    print("-- resume without fixing the cause")
    await process.resume()
    await ticket(4)

    print("-- raise the limit, then resume")
    process.config.budget.max_runs_per_day = 10
    process.apply_governance_config("budget")
    await process.resume()
    await ticket(5)
    print(f"state={process.state.value} runs={process.status()['invocation_count']}")
    await runtime.stop_all()
    await receiver.stop()
    await llm.stop()


asyncio.run(main())

POST /notifier {"event": "process.suspended", "process": "billing-bot", "reason": "budget"}
POST /slack {"text": ":rotating_light: Agent needs a human. billing-bot: max_runs_per_day hit (3/2)"}
POST /pagerduty {"routing_key": "local-test-key", "event_action": "trigger", "dedup_key": "billing-bot:budget_violation:max_runs_per_day", "payload": {"summary": "billing-bot: max_runs_per_day hit (3/2)", "source": "promptise-runtime", "severity": "error", "custom_details": {"type": "budget_violation", "process_id": "5b7aae6b-6e4d-4718-9458-83b47c937eb7", "process_name": "billing-bot", "violation": {"limit_name": "max_runs_per_day", "limit_value": 2, "current_value": 3, "tool_name": null}}}}
-- resume without fixing the cause
POST /notifier {"event": "process.suspended", "process": "billing-bot", "reason": "budget"}
POST /slack {"text": ":rotating_light: Agent needs a human. billing-bot: max_runs_per_day hit (4/2)"}
POST /pagerduty {"routing_key": "local-test-key", "event_action": "trigger", "dedup_key": "billing-bot:budget_violation:max_runs_per_day", "payload": {"summary": "billing-bot: max_runs_per_day hit (4/2)", "source": "promptise-runtime", "severity": "error", "custom_details": {"type": "budget_violation", "process_id": "5b7aae6b-6e4d-4718-9458-83b47c937eb7", "process_name": "billing-bot", "violation": {"limit_name": "max_runs_per_day", "limit_value": 2, "current_value": 4, "tool_name": null}}}}
-- raise the limit, then resume
state=running runs=3

The first three lines are the escalation path working: the notifier reported the suspension, the relay posted a Slack-shaped message, and a PagerDuty Events API v2 event with severity and the full payload under custom_details. In production you set SLACK_WEBHOOK_URL, PAGERDUTY_EVENTS_URL and PAGERDUTY_ROUTING_KEY from your secret store and change nothing else. Both URLs are secrets: Slack's documentation says so of webhook URLs, and the routing key lets anyone open incidents on your service.

The WebhookSink could deliver to the local receiver because it was created with allow_private_networks=True. Event notifier sinks have that option and a transform hook; escalation webhooks have neither.

The second half of the same output is the part teams get wrong:

-- resume without fixing the cause
POST /notifier {"event": "process.suspended", "process": "billing-bot", "reason": "budget"}
POST /slack {"text": ":rotating_light: Agent needs a human. billing-bot: max_runs_per_day hit (4/2)"}
POST /pagerduty {… "dedup_key": "billing-bot:budget_violation:max_runs_per_day", …}
-- raise the limit, then resume
state=running runs=3

Resuming without a fix let the next ticket hit the same limit, so the agent escalated and suspended again at once. The refused attempts count against the limit too: the counter went from 3 to 4. Because the dedup_key was the same, PagerDuty would add the repeat to the open incident instead of paging twice. Raising the limit with apply_governance_config("budget") and then resuming let the fifth ticket run.

The resume step for each check:

Events that arrived while the process was suspended wait in its queue and run after the resume. Pause and resume an AI agent covers suspension in detail.

escalation/
├── fake_llm.py                 # scripted model server, also answers as the mission judge
├── ops_bus.py                  # a minimal in-memory event bus
├── escalate_basics.py          # step 1: webhook, event type, no bus, bad URL
├── governance_escalations.py   # step 2: budget, health and mission escalations
├── receiver.py                 # a local stand-in for Slack and PagerDuty
├── on_call_relay.py            # step 3: bus subscriber that formats and delivers
└── route_to_pager.py           # steps 3-4: route, then resume two ways

pip install "promptise>=1.5.0"
python escalate_basics.py
python governance_escalations.py
python route_to_pager.py     # uses ports 8781 and 9785

Declare escalations next to the limits

In a manifest, escalation sits inside each governance section. With promptise runtime start, use webhook_url: the CLI's runtime has no event bus, so an event_type target there is dropped with a warning.

budget:
  enabled: true
  max_runs_per_day: 200
  on_exceeded: escalate
  escalation:
    webhook_url: https://relay.example.com/escalate/k3J9-long-random-path
health:
  enabled: true
  on_anomaly: escalate
  escalation:
    webhook_url: https://relay.example.com/escalate/k3J9-long-random-path

Put a relay you control in front of Slack and PagerDuty

The webhook is one POST with a 10-second timeout, no retry, no redirects and no signature. Point it at a small public relay, such as a serverless function, which formats the payload, retries, deduplicates and fans out. Or, when the runtime runs in your own Python program, subscribe a relay to the event bus as in step 3. That route can also reach internal systems such as an Alertmanager on a private network.

Protect the relay endpoint

Escalation requests carry no signature and no custom headers, so the URL is the only secret. Use a long random path, serve it over HTTPS, accept only JSON with the expected type values, and rate-limit it. Keep the URL out of version control, like Slack's own webhook URLs.

Keep pages rare and grouped

Health has a cooldown per anomaly, and a mission escalates once per pause. A budget escalates again on every refused event after a resume, so group by a dedup_key built from the process, the payload type and the limit or anomaly, as the relay does. Send warnings, recoveries and finished runs as notifications instead of pages.

Watch for escalations that went nowhere

Delivery failures never reach the agent: an SSRF refusal, an HTTP error status, a missing bus or a target with nothing set are all WARNING lines on promptise.runtime.escalation. Ship those logs and alert on them. Also, on_exceeded="escalate" without an escalation target suspends the process and tells nobody, and so does on_anomaly="escalate" without one.

Test the path before you need it

Run the governance checks with tiny limits and a fake model, as in step 2, and assert on what the relay received. escalate() can be called directly in a unit test with your bus. Exercise the real Slack and PagerDuty endpoints once in a staging service before relying on them.

Both send JSON when something happens, and both can reach Slack. They answer different questions. Notifications report: every event you subscribe to, to any number of sinks, with retries. Escalation acts: one message per incident from a governance check, together with a pause.

Many teams use both: notifications for the stream of what happened, escalation for the moments a person must act. AI agent notifications covers sinks, signing and transforms.

• Three sources only. Budget, health and mission escalate. A failed process (process.failed), exhausted restarts or dead-lettered events don't; send those through notifications.

• No delivery guarantees for the webhook: one attempt, no retry queue, no signature.

• Fixed payloads. You can't add fields such as a runbook link at the source; the relay adds them.

• No auto-resume. A suspended process waits for a person or your own automation.

• The event bus is bring-your-own, and the CLI runtime has none.

How do I make an AI agent escalate to a human?

Give each governance check a target. In Promptise, set on_exceeded="escalate" on the budget and on_anomaly="escalate" on health checks, and an escalation=EscalationTarget(...) on those and on the mission. When a limit, anomaly or unsure judge trips, the runtime sends the payload to your webhook or event bus and pauses the work.

When should an AI agent escalate?

When continuing could cost money, cause damage or waste time without progress: a spending or run limit is hit, the agent is stuck or looping, its replies are empty, its error rate is high, or a judge isn't confident the goal is getting closer. Routine events such as finished runs and warnings should be notifications, not escalations.

What does an escalation payload contain?

A type (budget_violation, health_anomaly or low_confidence), the process ID, and the facts behind it: the limit's name, value and the value that crossed it, the anomaly type and its details, or the judge's confidence and reasoning. Budget and health payloads also carry the process name.

Can I send AI agent escalations straight to Slack or PagerDuty?

Not with the raw payload: Slack needs a text field, and PagerDuty's Events API v2 needs a routing key, an action and a summary. Put a small relay in between, either a public endpoint the webhook calls or an event-bus subscriber, that formats the message, retries and deduplicates.

What is the difference between escalation and notifications?

Notifications report events to any number of sinks, with retries and reshaping, and change nothing. Escalation is a single message from a budget, health or mission check, and it comes with a pause: budget and health suspend the process, and a mission stops being judged until resumed.

What happens to the agent after it escalates?

A budget or health escalation suspends the process: its triggers keep accepting events, which wait in the queue, and no new run starts. A mission escalation pauses the mission while the process keeps running. Nothing resumes on its own; a person fixes the cause and resumes it from code, the API or the dashboard.

• The limits that trigger budget escalations: AI agent cost control.

• The anomaly detectors behind health escalations: AI agent health monitoring.

• Missions and the LLM judge: goal-driven AI agents.

• The event stream next to escalations: AI agent notifications to webhooks and Slack.

• Run hooks on refusals too: AI agent lifecycle hooks.

• Reference: budget escalation and health monitoring in the Promptise docs.