Event-driven agents: triggering AI from SAP business events

#The 60-second version

Most AI pilots on SAP data start on a schedule: every hour, a job asks SAP "what's blocked now?" and hands the list to an agent. That works, but it is slow and wasteful. The order blocked at 9:01 waits until 10:00, and most of the hourly checks find nothing new.

The alternative is to let SAP tell you. SAP S/4HANA can publish business events, short messages such as "sales order 4711 was changed". An event broker delivers each one to whoever subscribed. Your agent starts work within seconds, only when something happened.

The catch is that event delivery is messy by design. The same event can arrive twice. Events can arrive late or out of order. Some will be broken. A good event-driven agent is built to shrug all of that off: it ignores repeats, re-reads the current state from SAP instead of trusting the message, and parks bad events in a separate queue for a person to look at. Get that right and "react in seconds" is safe. Get it wrong and the agent proposes the same credit release three times.

#Why it matters to the business

Stay with the credit team's blocked sales orders from earlier Unit 9 topics.

• Speed where it counts. A blocked order for a key customer is a delivery that isn't moving. Starting triage when the block happens, not at the next batch run, gives the credit analyst a prepared case while the customer is still on the phone.
• Less load on SAP. A poll asks SAP the same question over and over. An event costs one message when something changes, plus one read to fetch the details.
• Duplicates cost money and trust. If a repeated event makes the agent send a second email to a customer or open a second approval, people stop trusting it. Handling repeats is not a technical nicety; it is the difference between a helpful assistant and a noisy one.
• Broken events are a support process. Somebody must own the queue of events the system could not process, with a target for how fast they are looked at. Otherwise orders silently go untriaged.
• Clean core. Events let you build the agent next to SAP, not inside it. SAP's architecture guidance presents event-based integration as a way for extensions to follow clean core principles.

#How SAP does it

As of October 2026, the pieces are spread across S/4HANA and SAP Business Technology Platform (BTP). State them carefully with your SAP account team, because names and packaging change.

• The events. S/4HANA publishes standard business events with names like sap.s4.beh.salesorder.v1.SalesOrder.Changed.v1. SAP lists its business events on the SAP Business Accelerator Hub. Coverage differs by scenario: SAP's training for one sales order replication scenario notes that deletions aren't supported there. Check the catalog for the object and release you have.
• SAP Cloud Application Event Hub gives access to business events from SAP cloud applications such as SAP S/4HANA Cloud and SAP SuccessFactors. SAP's CAP documentation calls it the new default offering for messaging on SAP BTP. Its tutorial is marked as needing a customer or partner license.
• The Event Mesh capability of SAP Integration Suite is positioned by SAP for starter or small-volume event scenarios.
• SAP Integration Suite, advanced event mesh is SAP's broker for large, distributed setups across SAP and non-SAP systems.
• SAP Build Process Automation can start processes from events. SAP's "Manage Credit Block on Sales Order" content package lists the sales order Created, Changed and Deleted events and an Event Mesh queue, although its guide says the package itself starts processes from a payload. Joule agents are a newer route; the SAP pages we opened in this run don't describe starting a Joule agent directly from a business event, so plan on an event reaching a process or your own service, which then calls the agent.

#Polling or events: a decision guide

Events are not always the right answer. Use this table for each trigger in your use case.

Many real designs use both: events for speed, plus a daily poll that catches anything an event missed.

#Questions to ask

• Which business events does our system publish for this object, in our release, and which are switched on today?
• Which broker do we already run, and who owns it?
• What happens if the same event arrives twice? Show me a test.
• Does the agent trust the event's content, or does it re-read the current state from SAP?
• Where do events go that can't be processed, who looks at them, and how fast?
• If the broker or our service is down for an hour, what is lost and what catches up?
• Which user or technical identity does the agent use when it reads SAP after an event?
• Do we still need a scheduled check as a safety net?

#Common misconceptions

• "Each event arrives exactly once." Brokers that guarantee delivery keep an event until it is acknowledged. If your service crashes before it acknowledges, the event comes again. Plan for repeats.
• "The event contains everything the agent needs." Treat the event as a notification. Read the current state from SAP before acting, because the order may have changed since.
• "Events arrive in the order they happened." Retries and parallel consumers can reorder them. Re-reading the current state makes the order of arrival matter much less.
• "Event-driven means no person is involved." The trigger is automatic. The approval rules from Designing human approval into AI workflows still apply to whatever the agent proposes.
• "Failed events can be retried forever." An event that is broken will fail every time. After a few tries it must move aside to a dead-letter queue so it doesn't block the rest.

#Key terms

All course terms, A to Z, in the glossary

• Business event: a message from a business system saying something happened to a business object, such as "sales order changed".
• Event broker: the service that receives events from publishers and delivers them to subscribers.
• Publish/subscribe: publishers send events to topics; subscribers say which topics they want.
• Queue: a place in the broker that holds events for one consumer until it confirms it has processed them.
• At-least-once delivery: every event is delivered, but some may be delivered more than once.
• Idempotent: doing something twice has the same effect as doing it once.
• Dead-letter queue: a separate queue for events that failed too many times, kept for a person to inspect.
• CloudEvents: an open standard for the envelope of an event (its ID, source, type and time).

#Check yourself

#Mental model

An event is a doorbell, not a parcel. It tells you that something happened to a business object, and which one. It may ring twice, ring late, or ring after a later ring. So your consumer does four things on every ring, in this order:

1. Check the envelope. Is this a well-formed event of a type we handle? If not, it's poison: don't retry it forever.
2. Have we answered this ring before? Look up the event's source and id. If yes, acknowledge and stop.
3. Open the door and look. Read the object's current state from SAP. Don't trust the event's contents for decisions.
4. Act once per fact, then acknowledge. Run the agent only if the facts it uses are new, store the result and the event ID in one transaction, and only then tell the broker you're done.

The rest of this topic is how each step works, why the order matters, and what SAP offers for the broker side. The agent itself is the triage agent from Agents from first principles and Multi-step agents; its proposals flow into the approval gate from Designing human approval into AI workflows.

#How it works

#Polling versus events

A poll asks "what's blocked?" on a timer. It's simple, needs no broker, and gives you a full snapshot every run. It's also late by up to one interval, and it reads unchanged data again and again.

With events, the system of record publishes a message when a business object changes. A broker routes it by topic to every subscriber. Subscribers usually read from a queue that the broker fills from one or more topics, so events wait while the consumer is offline.

flowchart LR
  S[S/4HANA<br/>sales order changed] -->|business event| B[Event broker<br/>topic]
  B --> Q[Queue for<br/>triage service]
  Q --> C[Consumer]
  C -->|read current state| API[SAP read API]
  C --> A[Triage agent]
  A --> P[Proposal]
  P --> G[Approval gate]
  C -. fails too often .-> D[Dead-letter queue]

#The envelope: CloudEvents

SAP's Event Hub tutorial says events must follow CloudEvents, an open specification hosted by the Cloud Native Computing Foundation. Four attributes are required:

Optional attributes include time, subject and datacontenttype. The payload goes in data. The specification says producers must make source plus id unique for each distinct event, that a re-sent duplicate may keep the same id, and that consumers may treat events with the same source and id as duplicates. That pair is your deduplication key.

What data contains depends on the event. Look up each event's payload on the SAP Business Accelerator Hub for your release. CAP's guide to S/4HANA events, for example, handles a business partner event by reading the business partner number from data. The lab's events carry only the sales order number, and the consumer reads everything else from SAP. That design works whatever the payload holds.

#Delivery guarantees and why duplicates happen

SAP's advanced event mesh documentation describes two modes. Direct messaging is fast but can drop messages, for example when no subscriber is connected. Guaranteed messaging stores the message on a durable endpoint, such as a queue, until each consumer acknowledges it.

Guaranteed delivery creates duplicates by design. Picture your consumer receiving an event, writing a proposal, and crashing before it sends the acknowledgement. The broker never heard "done", so it delivers the event again. This is at-least-once delivery. The cure is an idempotent consumer: processing the same event twice has the same effect as processing it once.

There are two layers of idempotency, and you want both:

• Message level: keep a table of processed (source, id) pairs. A repeat is acknowledged and skipped.
• Business level: two different events can mean the same thing. SAP may publish "changed" for an edit to the item text, which the triage agent doesn't care about. Compute a key from the order number and the facts the agent uses (block type, amounts, credit exposure). If a proposal already exists for that key, don't create another. If the facts changed, create a new proposal and mark the old one superseded, so an approver never approves stale data.

#The one transaction that makes it safe

The order of writes matters. In the lab, the proposal, the processed-event marker and the counters commit in one database transaction, and the acknowledgement goes to the broker after that commit.

sequenceDiagram
  participant B as Broker
  participant C as Consumer
  participant DB as Consumer database
  participant S as SAP
  B->>C: event (delivery 1)
  C->>DB: seen source+id before?
  DB-->>C: no
  C->>S: read current order
  C->>DB: BEGIN; insert proposal; insert source+id; COMMIT
  Note over C: crash here: no ack sent
  B->>C: same event (delivery 2)
  C->>DB: seen source+id before?
  DB-->>C: yes
  C->>B: ack

If the consumer crashes before the commit, nothing was stored and the redelivery processes the event normally. If it crashes after the commit, the redelivery finds the marker and skips. Either way there is exactly one proposal. If you sent an email or called SAP inside the handler, this guarantee would break, which is why the agent only proposes and the approval gate's execute step uses its own idempotency key.

#Ordering

Retries and parallel consumers can deliver an older event after a newer one. The doorbell model handles most of it: if you always read the current state, a late event just causes one extra read. Where order truly matters (for example, a stream of status changes that you store as history), compare the event's time with the last one you applied for that object, or process one object's events on one consumer. Don't build a design that depends on the broker preserving order unless your broker documents it for your setup.

#Retries and the dead-letter queue

Separate two kinds of failure:

• Transient: SAP timed out, the network blipped. Leave the event unacknowledged (or reject it so it's redelivered) and try again.
• Poison: the event is malformed or names an object that doesn't exist. It will fail every time.

Brokers solve the second with a maximum redelivery count. Solace's article on its PubSub+ broker describes moving a message that exceeds it to a dead message queue (also called a dead-letter queue), where it waits for special handling instead of blocking the queue. Someone fixes the cause and replays the event. Your consumer can also decide early: if the handler knows an event is poison, it can count it as failed straight away. Whatever you choose, monitor the dead-letter queue's depth and age; an unwatched one is where orders go to be forgotten.

#Which identity reads SAP

An event has no user behind it, so the agent runs under its own technical identity. Agent permissions and SAP authorizations covers this pattern: read-only, scoped to the objects and sales organizations it needs, owned by a named person. The event decides when the agent runs; it never widens what the agent may do.

#Build it yourself: an event-driven trigger for the blocked-orders triage

You will build unit09/event_agent.py. It contains a tiny local stand-in for an event broker, a queue of nine SAP-shaped events, and a consumer that triggers the triage agent safely. Then you'll make it crash mid-run, simulate an SAP timeout, send it a broken event, and replay that event from the dead-letter queue, checking at each step that no order gets a second proposal.

flowchart LR
  P[publish --sample] --> Q[(broker.db<br/>queue)]
  Q --> C[consume]
  C -->|validate, dedupe,<br/>read SAP, triage| A[(agent.db<br/>proposals)]
  C -->|3 failures| D[(broker.db<br/>dead-letter queue)]
  D -->|replay| Q

Before you start: complete Set up your computer for this course and Set up for Unit 9, which creates the unit09 folder. This lab uses only built-in Python modules, so there's nothing new to install. No SAP system or broker account is needed.

#What you need

• Your course folder orchestrate-course with its .venv, from the setup topics.
• About 40 minutes.
• No account and no cost. Everything runs offline on made-up data.

#Step 1: Open your course folder and turn on the virtual environment

1. Open VS Code, choose File > Open Folder, and open orchestrate-course.

2. Open a terminal: Terminal > New Terminal.

3. If the prompt doesn't start with (.venv), turn it on:

   • Windows (PowerShell):

     Copy

     .venv\Scripts\Activate.ps1

   • macOS / Linux:

     Copy

     source .venv/bin/activate

4. Check that the unit09 folder exists (same command on every system):

   Copy

   python -c "import pathlib; print(pathlib.Path('unit09').is_dir())"

   You should see True. If you see False, right-click in VS Code's file list, choose New Folder and name it unit09.

#Step 2: Save the script

1. In VS Code's file list, right-click unit09, choose New File and name it event_agent.py.
2. Paste the code below and save. The table after Step 9 explains each part.

"""Unit 9: an event-driven trigger for the blocked-orders triage agent.

A small local "broker" holds SAP-shaped business events (CloudEvents format). A consumer takes them one
at a time and, for each event: checks it is well formed, skips duplicates, re-reads the order's current
state, runs the triage agent only if the order is blocked and its facts are new, stores the proposal,
and only then acknowledges the event. Bad events are retried and then moved to a dead-letter queue.

Commands (run from your course folder, with .venv turned on):
    python unit09/event_agent.py reset                     # start clean (deletes this lab's files)
    python unit09/event_agent.py publish --sample          # put nine made-up events on the queue
    python unit09/event_agent.py consume                   # process everything on the queue
    python unit09/event_agent.py consume --crash-after 2   # stop after 2 proposals, before acknowledging
    python unit09/event_agent.py consume --flaky-sap       # the first SAP read for order 4713 times out
    python unit09/event_agent.py status                    # queue, dead-letter queue, proposals, counters
    python unit09/event_agent.py dlq                       # show dead-lettered events and why
    python unit09/event_agent.py replay --seq 7            # move one dead-lettered event back to the queue
    python unit09/event_agent.py publish --order 4711      # a fresh "Changed" event for one order

All data is made up. Nothing here reaches an SAP system. Built-in Python modules only.
"""
import argparse
import hashlib
import json
import re
import sqlite3
import sys
import uuid
from datetime import datetime, timedelta, timezone
from pathlib import Path

HERE = Path(__file__).resolve().parent
LAB = HERE / "event_lab"
BROKER_DB = LAB / "broker.db"   # stands in for the event broker: a queue and a dead-letter queue
AGENT_DB = LAB / "agent.db"     # belongs to your app: processed event IDs and proposals
SAP_STUB = LAB / "sap_orders.json"  # stands in for SAP: the current state of each order

MAX_DELIVERIES = 3   # after this many failed deliveries an event goes to the dead-letter queue
SOURCE = "/default/sap.s4.beh/S4DCLNT100"   # made-up source system; yours comes from your channel setup
TYPES = {
    "sap.s4.beh.salesorder.v1.SalesOrder.Created.v1",
    "sap.s4.beh.salesorder.v1.SalesOrder.Changed.v1",
}

# Made-up orders in sales organization 1010, shaped like API_SALES_ORDER_SRV headers plus block facts.
SAMPLE_ORDERS = {
    "4711": {"SalesOrder": "4711", "SoldToParty": "17100001", "TotalNetAmount": 41200.00,
             "TransactionCurrency": "EUR", "block": "credit", "credit_limit": 400000.00,
             "exposure_after_order": 412000.00, "open_overdue": 0.00},
    "4712": {"SalesOrder": "4712", "SoldToParty": "17100002", "TotalNetAmount": 88000.00,
             "TransactionCurrency": "EUR", "block": "credit", "credit_limit": 250000.00,
             "exposure_after_order": 295000.00, "open_overdue": 20000.00},
    "4713": {"SalesOrder": "4713", "SoldToParty": "17100003", "TotalNetAmount": 2300.00,
             "TransactionCurrency": "EUR", "block": "delivery", "note": "ship-to address has no postal code"},
    "4714": {"SalesOrder": "4714", "SoldToParty": "17100004", "TotalNetAmount": 15600.00,
             "TransactionCurrency": "EUR", "block": "billing", "note": "billing block set manually; no text"},
    "4715": {"SalesOrder": "4715", "SoldToParty": "17100005", "TotalNetAmount": 950.00,
             "TransactionCurrency": "EUR", "block": "delivery",
             "note": "same ship-to issue as last week; owner unclear"},
    "4716": {"SalesOrder": "4716", "SoldToParty": "17100006", "TotalNetAmount": 5100.00,
             "TransactionCurrency": "EUR", "block": None},
}


# ---------- small helpers ----------
def now() -> datetime:
    return datetime.now(timezone.utc).replace(microsecond=0)


def connect(path: Path) -> sqlite3.Connection:
    LAB.mkdir(exist_ok=True)
    con = sqlite3.connect(path)
    con.row_factory = sqlite3.Row
    return con


def broker() -> sqlite3.Connection:
    con = connect(BROKER_DB)
    con.execute("CREATE TABLE IF NOT EXISTS queue (seq INTEGER PRIMARY KEY AUTOINCREMENT, body TEXT, "
                "deliveries INTEGER DEFAULT 0, leased INTEGER DEFAULT 0, last_error TEXT DEFAULT '')")
    con.execute("CREATE TABLE IF NOT EXISTS dlq (seq INTEGER PRIMARY KEY, body TEXT, deliveries INTEGER, "
                "last_error TEXT, moved_at TEXT)")
    return con


def agent_db() -> sqlite3.Connection:
    con = connect(AGENT_DB)
    con.execute("CREATE TABLE IF NOT EXISTS processed (source TEXT, id TEXT, outcome TEXT, at TEXT, "
                "PRIMARY KEY (source, id))")
    con.execute("CREATE TABLE IF NOT EXISTS proposals (business_key TEXT PRIMARY KEY, sales_order TEXT, "
                "kind TEXT, recommendation TEXT, evidence TEXT, from_event TEXT, at TEXT, state TEXT)")
    con.execute("CREATE TABLE IF NOT EXISTS counters (name TEXT PRIMARY KEY, value INTEGER)")
    return con


def bump(con: sqlite3.Connection, name: str) -> None:
    con.execute("INSERT INTO counters VALUES (?, 1) ON CONFLICT(name) DO UPDATE SET value = value + 1",
                (name,))


# ---------- the publishing side (stands in for SAP S/4HANA and the broker) ----------
def make_event(event_type: str, data: dict, minutes_ago: int, event_id: str = "") -> dict:
    """A CloudEvents 1.0 envelope. The data carries only the key: the consumer reads the rest from SAP."""
    return {"specversion": "1.0", "type": event_type, "source": SOURCE,
            "id": event_id or str(uuid.uuid4()),
            "time": (now() - timedelta(minutes=minutes_ago)).isoformat(),
            "datacontenttype": "application/json", "data": data}


def sample_events() -> list:
    """Nine events, including the awkward cases a real broker will send you sooner or later."""
    changed = "sap.s4.beh.salesorder.v1.SalesOrder.Changed.v1"
    created = "sap.s4.beh.salesorder.v1.SalesOrder.Created.v1"
    first = make_event(created, {"SalesOrder": "4711"}, 9, "a1f0c3e2-7d41-4b7e-9c11-5e2a00004711")
    return [
        first,  # 1: a new order that is credit blocked
        make_event(changed, {"SalesOrder": "4712"}, 8, "b2e1d4f3-8e52-4c8f-8d22-6f3b00004712"),
        make_event(changed, {"SalesOrder": "4713"}, 7, "c3d2e5a4-9f63-4d90-9e33-7a4c00004713"),
        dict(first),  # 4: the broker delivers event 1 a second time (same id)
        # 5: order 4711 changed again, but nothing the triage uses moved
        make_event(changed, {"SalesOrder": "4711"}, 5, "d4c3f6b5-a074-4ea1-8f44-8b5d00004711"),
        # 6: an order that isn't blocked at all
        make_event(changed, {"SalesOrder": "4716"}, 4, "e5b4a7c6-b185-4fb2-9a55-9c6e00004716"),
        # 7: malformed, the key is in the wrong field
        make_event(changed, {"Order": "4714"}, 3, "f6a5b8d7-c296-40c3-8b66-ad7f00004714"),
        # 8: arrives late: it is older than events 2 to 7
        make_event(changed, {"SalesOrder": "4714"}, 12, "07f6c9e8-d3a7-41d4-9c77-be8000004714"),
        make_event(changed, {"SalesOrder": "4715"}, 2, "18e7daf9-e4b8-42e5-8d88-cf9100004715"),
    ]


def publish(events: list) -> None:
    con = broker()
    with con:
        for e in events:
            con.execute("INSERT INTO queue (body) VALUES (?)", (json.dumps(e),))
    print(f"Published {len(events)} event(s). Queue depth is now "
          f"{con.execute('SELECT COUNT(*) FROM queue').fetchone()[0]}.")


def receive(con: sqlite3.Connection):
    """Give out the oldest unleased event. Events that failed too often go to the dead-letter queue."""
    while True:
        row = con.execute("SELECT * FROM queue WHERE leased = 0 ORDER BY seq LIMIT 1").fetchone()
        if row is None:
            return None
        if row["deliveries"] >= MAX_DELIVERIES:
            with con:
                con.execute("INSERT INTO dlq VALUES (?, ?, ?, ?, ?)",
                            (row["seq"], row["body"], row["deliveries"], row["last_error"],
                             now().isoformat()))
                con.execute("DELETE FROM queue WHERE seq = ?", (row["seq"],))
            print(f"  seq {row['seq']}: moved to the dead-letter queue after "
                  f"{row['deliveries']} failed deliveries")
            continue
        with con:
            con.execute("UPDATE queue SET deliveries = deliveries + 1, leased = 1 WHERE seq = ?",
                        (row["seq"],))
        return row["seq"], json.loads(row["body"]), row["deliveries"] + 1


def ack(con: sqlite3.Connection, seq: int) -> None:
    with con:
        con.execute("DELETE FROM queue WHERE seq = ?", (seq,))


def nack(con: sqlite3.Connection, seq: int, error: str) -> None:
    with con:
        con.execute("UPDATE queue SET leased = 0, last_error = ? WHERE seq = ?", (error, seq))


# ---------- the consuming side (your app) ----------
class Poison(Exception):
    """The event itself is wrong. Retrying won't help; it will end in the dead-letter queue."""


def validate(event: dict) -> str:
    """Check the CloudEvents attributes we rely on and return the sales order number."""
    for attr in ("specversion", "id", "source", "type"):
        if not event.get(attr):
            raise Poison(f"missing CloudEvents attribute '{attr}'")
    if event["type"] not in TYPES:
        raise Poison(f"unexpected event type {event['type']}")
    order = (event.get("data") or {}).get("SalesOrder")
    if not order:
        raise Poison("data has no SalesOrder")
    # Event data is untrusted: accept only a plain document number before it goes near a query or URL.
    if not isinstance(order, str) or not re.fullmatch(r"[0-9]{1,10}", order):
        raise Poison("SalesOrder is not a document number of up to 10 digits")
    return order


def read_order(order: str, flaky: bool, con: sqlite3.Connection) -> dict:
    """Stands in for a read API call (for example API_SALES_ORDER_SRV). Always reads the CURRENT state."""
    if flaky and order == "4713":
        tries = con.execute("SELECT value FROM counters WHERE name = 'sap_reads_4713'").fetchone()
        with con:
            bump(con, "sap_reads_4713")
        if tries is None:
            raise TimeoutError("SAP read timed out (simulated)")
    orders = json.loads(SAP_STUB.read_text(encoding="utf-8"))
    if order not in orders:
        raise Poison(f"order {order} not found in SAP")
    return orders[order]


def triage(order: dict) -> dict:
    """Rule-based stand-in for the triage agent from earlier Unit 9 topics. It proposes; it never acts."""
    if order["block"] == "credit":
        over = order["exposure_after_order"] - order["credit_limit"]
        pct = over / order["credit_limit"] * 100
        return {"kind": "request_credit_release", "recommendation": "release" if order["open_overdue"] == 0
                else "hold", "evidence": [f"{pct:.1f}% over limit", f"overdue {order['open_overdue']:,.0f}"]}
    if "owner unclear" in order.get("note", ""):
        return {"kind": "assign_worklist_owner", "recommendation": "assign", "evidence": [order["note"]]}
    if "postal code" in order.get("note", ""):
        return {"kind": "draft_customer_email", "recommendation": "send", "evidence": [order["note"]]}
    return {"kind": "recommend_next_check", "recommendation": "check", "evidence": [order.get("note", "")]}


def business_key(order: dict) -> str:
    """Same order + same facts = same key. A repeat or irrelevant change can't create a second proposal."""
    facts = {k: order.get(k) for k in ("block", "TotalNetAmount", "credit_limit",
                                        "exposure_after_order", "open_overdue", "note")}
    digest = hashlib.sha256(json.dumps(facts, sort_keys=True).encode()).hexdigest()[:12]
    return f"{order['SalesOrder']}:{order['block']}:{digest}"


def handle(event: dict, con: sqlite3.Connection, flaky: bool) -> str:
    """Process one event. Everything the event changes is written in ONE transaction with its ID."""
    order_no = validate(event)
    if con.execute("SELECT 1 FROM processed WHERE source = ? AND id = ?",
                   (event["source"], event["id"])).fetchone():
        with con:
            bump(con, "duplicates_skipped")
        return f"duplicate of event {event['id'][:8]}, skipped"
    order = read_order(order_no, flaky, con)
    outcome = ""
    with con:  # one transaction: proposal (if any), processed marker and counters commit together
        if not order.get("block"):
            outcome = f"order {order_no} is not blocked, no action"
            bump(con, "not_blocked")
        else:
            key = business_key(order)
            if con.execute("SELECT 1 FROM proposals WHERE business_key = ?", (key,)).fetchone():
                outcome = f"order {order_no} already has a proposal for these facts, no new one"
                bump(con, "no_new_facts")
            else:
                p = triage(order)
                older = con.execute("UPDATE proposals SET state = 'superseded' WHERE sales_order = ? "
                                    "AND state = 'open'", (order_no,)).rowcount
                con.execute("INSERT INTO proposals VALUES (?, ?, ?, ?, ?, ?, ?, 'open')",
                            (key, order_no, p["kind"], p["recommendation"], json.dumps(p["evidence"]),
                             event["id"], now().isoformat()))
                outcome = f"order {order_no}: proposal {p['kind']} ({p['recommendation']})"
                if older:
                    outcome += f", replaces {older} older proposal(s)"
                bump(con, "proposals")
        con.execute("INSERT INTO processed VALUES (?, ?, ?, ?)",
                    (event["source"], event["id"], outcome, now().isoformat()))
    return outcome


def consume(crash_after: int, flaky: bool) -> None:
    b, a = broker(), agent_db()
    with b:  # a new consumer: the broker hands back anything a crashed consumer never acknowledged
        released = b.execute("UPDATE queue SET leased = 0 WHERE leased = 1").rowcount
    if released:
        print(f"Broker: {released} unacknowledged event(s) from an earlier run are delivered again.")
    made = 0
    while True:
        got = receive(b)
        if got is None:
            break
        seq, event, delivery = got
        label = f"seq {seq} (delivery {delivery})"
        try:
            outcome = handle(event, a, flaky)
        except Poison as e:
            nack(b, seq, f"poison: {e}")
            print(f"  {label}: REJECTED, {e}")
            continue
        except Exception as e:  # transient: timeouts, network errors. Retry later.
            nack(b, seq, f"{type(e).__name__}: {e}")
            print(f"  {label}: RETRY, {type(e).__name__}: {e}")
            continue
        print(f"  {label}: {outcome}")
        if "proposal " in outcome:
            made += 1
            if crash_after and made >= crash_after:
                print(f"Simulated crash after {made} proposal(s): the last event was processed "
                      f"but never acknowledged.")
                sys.exit(1)
        ack(b, seq)
    print("Queue empty.")


# ---------- inspection ----------
def status() -> None:
    b, a = broker(), agent_db()
    q = b.execute("SELECT COUNT(*) FROM queue").fetchone()[0]
    d = b.execute("SELECT COUNT(*) FROM dlq").fetchone()[0]
    print(f"Queue: {q} waiting    Dead-letter queue: {d}")
    counters = {r["name"]: r["value"] for r in a.execute("SELECT * FROM counters")}
    for name in ("proposals", "duplicates_skipped", "no_new_facts", "not_blocked"):
        print(f"  {name:<20} {counters.get(name, 0)}")
    print("Proposals:")
    rows = a.execute("SELECT * FROM proposals ORDER BY rowid").fetchall()
    for r in rows:
        print(f"  {r['sales_order']}  {r['state']:<10} {r['kind']:<22} {r['recommendation']:<7} "
              f"{', '.join(json.loads(r['evidence']))}")
    if not rows:
        print("  (none yet)")


def show_dlq() -> None:
    rows = broker().execute("SELECT * FROM dlq ORDER BY seq").fetchall()
    if not rows:
        print("Dead-letter queue is empty.")
    for r in rows:
        e = json.loads(r["body"])
        print(f"seq {r['seq']}: {e.get('type')} id={e.get('id', '')[:8]} data={json.dumps(e.get('data'))}")
        print(f"   deliveries={r['deliveries']} last error: {r['last_error']}")


def replay(seq: int) -> None:
    b = broker()
    row = b.execute("SELECT * FROM dlq WHERE seq = ?", (seq,)).fetchone()
    if row is None:
        sys.exit(f"No dead-lettered event with seq {seq}. Run: python unit09/event_agent.py dlq")
    body = json.loads(row["body"])
    if not (body.get("data") or {}).get("SalesOrder") and "Order" in (body.get("data") or {}):
        body["data"] = {"SalesOrder": body["data"]["Order"]}  # the lab's "fix" for event 7
        print(f"Fixed the payload of seq {seq}: data.Order renamed to data.SalesOrder.")
    with b:
        b.execute("INSERT INTO queue (body) VALUES (?)", (json.dumps(body),))
        b.execute("DELETE FROM dlq WHERE seq = ?", (seq,))
    print(f"Moved seq {seq} back to the queue. Run consume to process it.")


def reset() -> None:
    for p in (BROKER_DB, AGENT_DB, SAP_STUB):
        if p.exists():
            p.unlink()
    LAB.mkdir(exist_ok=True)
    SAP_STUB.write_text(json.dumps(SAMPLE_ORDERS, indent=2), encoding="utf-8")
    print(f"Clean start. Made-up SAP orders written to {SAP_STUB.relative_to(HERE.parent)}")


def main() -> None:
    ap = argparse.ArgumentParser(description="Event-driven trigger for the blocked-orders triage agent.")
    sub = ap.add_subparsers(dest="cmd", required=True)
    sub.add_parser("reset")
    p = sub.add_parser("publish")
    p.add_argument("--sample", action="store_true", help="publish the nine sample events")
    p.add_argument("--order", help="publish one fresh Changed event for this order")
    c = sub.add_parser("consume")
    c.add_argument("--crash-after", type=int, default=0, help="stop after N proposals, before acknowledging")
    c.add_argument("--flaky-sap", action="store_true", help="the first SAP read for order 4713 times out")
    sub.add_parser("status")
    sub.add_parser("dlq")
    r = sub.add_parser("replay")
    r.add_argument("--seq", type=int, required=True)
    args = ap.parse_args()

    if args.cmd != "reset" and not SAP_STUB.exists():
        sys.exit("No lab data yet. Run first: python unit09/event_agent.py reset")
    if args.cmd == "reset":
        reset()
    elif args.cmd == "publish":
        if args.sample:
            publish(sample_events())
        elif args.order:
            changed = "sap.s4.beh.salesorder.v1.SalesOrder.Changed.v1"
            publish([make_event(changed, {"SalesOrder": args.order}, 0)])
        else:
            sys.exit("Say what to publish: --sample or --order 4711")
    elif args.cmd == "consume":
        consume(args.crash_after, args.flaky_sap)
    elif args.cmd == "status":
        status()
    elif args.cmd == "dlq":
        show_dlq()
    elif args.cmd == "replay":
        replay(args.seq)


if __name__ == "__main__":
    main()

#Step 3: Start clean and publish the sample events

1. Run (same on every system):

   Copy

   python unit09/event_agent.py reset
   python unit09/event_agent.py publish --sample

2. You should see:

   Copy

   Clean start. Made-up SAP orders written to unit09/event_lab/sap_orders.json
   Published 9 event(s). Queue depth is now 9.

   On Windows the path prints with backslashes. Open unit09/event_lab/sap_orders.json to see the six made-up orders. Order 4716 has "block": null: it isn't blocked.

The nine events include the awkward cases on purpose: event 4 is event 1 delivered again, event 5 is a change to 4711 that doesn't touch anything the triage uses, event 6 is for an unblocked order, event 7 has the order number in the wrong field, and event 8 is older than the events before it.

#Step 4: Crash the consumer halfway

1. Run:

   Copy

   python unit09/event_agent.py consume --crash-after 2

2. You should see:

   Copy

     seq 1 (delivery 1): order 4711: proposal request_credit_release (release)
     seq 2 (delivery 1): order 4712: proposal request_credit_release (hold)
   Simulated crash after 2 proposal(s): the last event was processed but never acknowledged.

The proposal for 4712 is saved, but the broker never got the acknowledgement for event 2. In a real broker, this is what happens when your service is killed between "work done" and "ack sent".

#Step 5: Restart with a flaky SAP connection

1. Run:

   Copy

   python unit09/event_agent.py consume --flaky-sap

2. You should see:

   Copy

   Broker: 1 unacknowledged event(s) from an earlier run are delivered again.
     seq 2 (delivery 2): duplicate of event b2e1d4f3, skipped
     seq 3 (delivery 1): RETRY, TimeoutError: SAP read timed out (simulated)
     seq 3 (delivery 2): order 4713: proposal draft_customer_email (send)
     seq 4 (delivery 1): duplicate of event a1f0c3e2, skipped
     seq 5 (delivery 1): order 4711 already has a proposal for these facts, no new one
     seq 6 (delivery 1): order 4716 is not blocked, no action
     seq 7 (delivery 1): REJECTED, data has no SalesOrder
     seq 7 (delivery 2): REJECTED, data has no SalesOrder
     seq 7 (delivery 3): REJECTED, data has no SalesOrder
     seq 7: moved to the dead-letter queue after 3 failed deliveries
     seq 8 (delivery 1): order 4714: proposal recommend_next_check (check)
     seq 9 (delivery 1): order 4715: proposal assign_worklist_owner (assign)
   Queue empty.

Read it line by line. Event 2 came back and was skipped by its ID: no second proposal for 4712. The timeout on 4713 was retried and then worked. Event 4 (the repeat of event 1) was skipped by its ID. Event 5 was a new event, but the business key showed nothing had changed for 4711. Event 7 failed three times and moved aside. Event 8 arrived late and didn't matter, because the consumer read 4714's current state.

#Step 6: Check the status and the dead-letter queue

1. Run:

   Copy

   python unit09/event_agent.py status
   python unit09/event_agent.py dlq

2. You should see:

   Copy

   Queue: 0 waiting    Dead-letter queue: 1
     proposals            5
     duplicates_skipped   2
     no_new_facts         1
     not_blocked          1
   Proposals:
     4711  open       request_credit_release release 3.0% over limit, overdue 0
     4712  open       request_credit_release hold    18.0% over limit, overdue 20,000
     4713  open       draft_customer_email   send    ship-to address has no postal code
     4714  open       recommend_next_check   check   billing block set manually; no text
     4715  open       assign_worklist_owner  assign  same ship-to issue as last week; owner unclear
   seq 7: sap.s4.beh.salesorder.v1.SalesOrder.Changed.v1 id=f6a5b8d7 data={"Order": "4714"}
      deliveries=3 last error: poison: data has no SalesOrder

Five blocked orders, five proposals, despite thirteen deliveries. That is the whole point of the lab.

#Step 7: Fix and replay the broken event

1. Run:

   Copy

   python unit09/event_agent.py replay --seq 7
   python unit09/event_agent.py consume

2. You should see:

   Copy

   Fixed the payload of seq 7: data.Order renamed to data.SalesOrder.
   Moved seq 7 back to the queue. Run consume to process it.
     seq 10 (delivery 1): order 4714 already has a proposal for these facts, no new one
   Queue empty.

The replayed event gets a new sequence number from the broker but keeps its CloudEvents id. It's processed now, and the business key stops it from duplicating the 4714 proposal. In real life the "fix" is a person correcting a mapping or a bug, not a script.

#Step 8: Change an order in "SAP" and send a new event

1. Open unit09/event_lab/sap_orders.json. In the "4711" block, change "open_overdue": 0.0 to "open_overdue": 5000.0 and save.

2. Run:

   Copy

   python unit09/event_agent.py publish --order 4711
   python unit09/event_agent.py consume
   python unit09/event_agent.py status

3. You should see order 4711: proposal request_credit_release (hold), replaces 1 older proposal(s), and in the status list the first 4711 proposal is now superseded with a new open one at the bottom recommending hold.

The facts changed, so the business key changed, and the old proposal is marked superseded so nobody approves it.

#Step 9: Save your work in Git

1. Check what Git sees:

   Copy

   git status

2. Commit the script but not the lab data. The event_lab folder holds test databases that you can recreate with reset. Add it to .gitignore first:

   • Windows (PowerShell):

     Copy

     Add-Content .gitignore "unit09/event_lab/"

   • macOS / Linux:

     Copy

     echo "unit09/event_lab/" >> .gitignore

3. Then commit:

   Copy

   git add .gitignore unit09/event_agent.py
   git commit -m "Unit 9: event-driven trigger for the triage agent"

#What each part of the script does

#If something goes wrong

The lab has no API keys, so there's no key to get wrong. The production sketch below needs broker credentials; they go in .env, never in code.

#The SAP way

#Getting events out of S/4HANA

S/4HANA's standard business events follow a naming pattern you've seen in the lab: sap.s4.beh.<object>.v1.<Object>.<Action>.v1. SAP's training for sales order replication lists sap.s4.beh.salesorder.v1.SalesOrder.Created.v1 and ...Changed.v1, and notes that sales order deletion and item deletion aren't supported in that scenario. SAP's "Manage Credit Block on Sales Order" process package guide lists Created, Changed and Deleted sales order topics, so availability differs by scenario and release. The SAP Business Accelerator Hub is where SAP publishes its business events; check the event and its payload for your edition and release.

SAP's resilient-apps sample sets up the connection inside the S/4HANA system with transaction /IWXBE/CONFIG: create a channel from an Event Mesh service key, activate it, check the connection, then add outbound bindings for the topics you want (the sample binds BusinessPartner/*). SAP's architecture guidance points to RAP for custom or derived events, and to the Application Interface Framework for older S/4HANA and ECC releases.

#Choosing a broker

As of October 2026, SAP's architecture guidance (last updated October 2025) describes three options:

SAP's Event Hub tutorial says events must follow CloudEvents and that the tutorial requires a customer or partner license. We didn't confirm a free trial plan for any of the three in this run, which is why the lab needs none.

#SAP Build Process Automation as the receiver

If your organization already approves work in SAP Build Process Automation, the event can start a process there, and the process calls your agent and routes its proposal to an approver. SAP's CodeJam on event-based processes forwards events from an advanced event mesh queue through a REST Delivery Point to SAP Build Process Automation, with the header Content-Type: application/cloudevents+json, and shows received events under Monitoring > Acquired Events > Business Events. For credit blocks, SAP's content package reads blocked orders through the Sales Document with Credit Block API (API_SLS_DOC_WITH_CREDIT_BLOCK, entity A_CreditBlockedSalesDocument) and supports S/4HANA Cloud Private Edition 2023 onwards and Public Edition 2502 onwards. That read is the "open the door and look" step from the mental model.

#Your own consumer (sketch)

The lab's handle function stays the same; only the transport changes. Here it's a webhook that a broker posts CloudEvents to:

# SKETCH: needs a broker that delivers to this URL, and FastAPI from earlier units.
import hmac
import os

from dotenv import load_dotenv
from fastapi import FastAPI, HTTPException, Request

from event_agent import Poison, agent_db, handle  # the lab code, unchanged

load_dotenv()
EXPECTED = os.environ["BROKER_WEBHOOK_TOKEN"]  # from .env or a secret store, never in code or logs
app = FastAPI()


@app.post("/events/sales-order")
async def on_event(request: Request):
    # Check the caller before reading the body. A shared bearer token stands in here; use what your
    # broker supports (OAuth token validation or a client certificate) per its setup.
    sent = request.headers.get("Authorization", "")
    if not hmac.compare_digest(sent.encode(), f"Bearer {EXPECTED}".encode()):
        raise HTTPException(status_code=401, detail="unauthorized")
    event = await request.json()
    try:
        outcome = handle(event, agent_db(), flaky=False)  # validate, dedupe, read SAP, store, all before replying
    except Poison as e:
        # Tell the broker this event can't be processed. How it treats each status code
        # (retry, discard, dead-letter) is broker configuration: check yours.
        raise HTTPException(status_code=400, detail=str(e))
    # Any other exception becomes a 500, which a broker typically treats as "try again".
    return {"status": "processed", "outcome": outcome}

Two production changes: the processed-events table moves to a shared database (so several instances see the same IDs, with the primary key catching races), and the SAP read uses the technical identity from Agent permissions and SAP authorizations.

#Joule agents

The SAP pages we opened in this run don't describe starting a Joule agent directly from an S/4HANA business event. Until SAP documents that, route events to a process or to your own consumer, and let it call the agent. Recheck this when Joule Studio's general availability is announced; see Joule and Joule Studio.

#Build vs. SAP

Whichever broker you choose, the consumer logic in this topic (validate, deduplicate, re-read, act once) is yours to write and test.

#Production concerns

• Security. Authenticate the broker to your endpoint (OAuth client credentials or certificates, as SAP's examples use) and reject everything else. Treat event data as untrusted input: validate types and lengths, and never pass free text from an event into a prompt as instructions. Prompt injection through a sales order text field is still prompt injection.
• Authorizations. The consumer reads SAP as a technical user scoped to the objects and organizations in the use case. An event never grants more access; if the agent proposes a change, the approval gate's execute step is a separate path.
• Evaluation. Add event-handling cases to your agent evaluation set from Evaluating agents: a repeat, a late event, a change that doesn't matter, a change that does, a poison event. Count proposals per order: more than one open proposal per order is a defect.
• Operations. Watch queue depth, age of the oldest message, dead-letter queue depth and age, redelivery rate, and time from event to proposal. Alert on dead-letter growth. Observability for AI systems in Unit 10 covers tracing; put the event id on every trace.
• Retention of processed IDs. The table grows forever unless you expire it. Keep IDs longer than the broker's longest redelivery window; the business key catches anything older.
• Cost. Each event that passes the filters costs one SAP read and, if the facts are new, one agent run. The business key keeps model calls proportional to real changes, not to event volume. Filter early: a "changed" event for an unblocked order should stop before the model is called.
• Backfill. Events start when you subscribe. Run a one-off poll to pick up orders that were already blocked, and keep a daily reconciliation poll as a safety net.
• Clean core. Use released events and released APIs. Custom events belong in RAP, not in modifications.

#Pitfalls

• Acknowledging before the work is saved. A crash then loses the event for good.
• Side effects inside the handler. Sending an email or changing SAP in the handler means a redelivery repeats it. Propose only; execute in a separate, idempotent step.
• Deduplicating by event ID only. Different events about the same unchanged facts create duplicate proposals. Add a business key.
• Trusting the payload. Acting on fields in the event instead of the current state acts on the past.
• Infinite retries. A poison event retried forever blocks the queue or burns money. Set a maximum and a dead-letter queue.
• Nobody owns the dead-letter queue. It fills quietly, and orders are never triaged.
• Forgetting the backlog. The agent only sees changes after go-live; blocked orders from before stay invisible.
• Assuming every change is covered. Check the event catalog and the scenario documentation for your release. Some changes, such as deletions in one SAP sales order scenario, aren't covered.

#Exercise

You'll add a new event type to the consumer, prove it doesn't create duplicates, and write a one-page event contract. The contract feeds the exception management capstone in Unit 14, where one platform handles exceptions from several processes.

1. Run python unit09/event_agent.py reset and python unit09/event_agent.py publish --sample, then python unit09/event_agent.py consume.
2. Open unit09/event_agent.py and find MAX_DELIVERIES = 3. Change it to MAX_DELIVERIES = 5, save, and run reset, publish --sample and consume again. Count the REJECTED lines for seq 7: there should be five. Change it back to 3.
3. In unit09/event_lab/sap_orders.json, change order 4716's "block": null to "block": "credit", and add "credit_limit": 100000.0, "exposure_after_order": 104000.0, "open_overdue": 0.0 inside the same block. Save. (Careful with commas: every value except the last in a block ends with a comma.)
4. Run python unit09/event_agent.py publish --order 4716 and python unit09/event_agent.py consume. You should see a new request_credit_release proposal for 4716.
5. Run the same two commands again. The second time you should see order 4716 already has a proposal for these facts, no new one.
6. Create unit09/event_contract.md with a table of two columns, item and decision, and these rows: events subscribed (names), filter (which orders the agent ignores), deduplication key, business key fields, retry limit, dead-letter owner and response time, SAP read API and technical user, safety-net poll schedule.
7. Commit: git add unit09/event_contract.md && git commit -m "Unit 9: event contract for the triage agent" (check git status first: no event_lab files).

Done when: step 2 shows five REJECTED lines for seq 7, step 4 creates a 4716 proposal, step 5 creates none, and event_contract.md has all eight rows filled in.

#Going further

For readers who want the specifications and broker features behind the patterns in this topic.

• CloudEvents specification. The core specification defines the required attributes and the rule that source plus id identifies an event, which is the basis for deduplication. It also defines the optional time and subject attributes, useful for ordering checks and for routing without opening the payload. CloudEvents spec
• Guaranteed versus direct delivery. SAP's advanced event mesh documentation explains why only guaranteed messaging survives a disconnected consumer, and how messages are removed only after every matching consumer acknowledges them. Read it to understand where your queue's durability comes from and what you give up with direct messaging. Message delivery modes
• Dead message queues in practice. Solace's article walks through the two ways a message loses delivery eligibility (time-to-live expiry and the maximum redelivery count), the queue settings each needs, and a detail that catches teams out: messages not flagged as eligible are deleted instead of moved. Check the equivalent setting in whichever broker you use. Solace on DMQs
• Exactly-once is a property of the whole system. No broker alone gives you "exactly once" side effects. You get the effect by combining at-least-once delivery, a deduplication store written in the same transaction as the result, and idempotent downstream calls. The lab's single transaction plus the approval gate's idempotency key is a small, complete example. The harder cases are when the result lives in a different system from the deduplication store; there you need an idempotency key the downstream system honors, or an outbox table that a separate process forwards.

#Check yourself