Example: a Slack approval surface

The approval loop's runtime half ships with beam-agents: an activation calls ctx.request_approval(...), suspends with a fail-closed deadline, and its kind = APPROVAL intent lands on the outbox; the effector routes it verbatim to --approvals-to and never executes it. What closes the loop is an AgentEnvelope.Approval arriving back on the pipeline's approvals topic, on the same key.

Nothing in the runtime fronts that channel — approval surfaces are yours, not the runtime's. This example is a worked one, so the contract is code you can read rather than a paragraph you re-derive.

Everything inside the dashed region is code you write; the runtime's half is the amber leg, .intents to the outbox and an effector that routes the request without executing it. The verdict returns as one AgentEnvelope.Approval on the approvals topic — the only hop here that is an edge in the Beam graph, and the re-injection that resumes the suspended activation on the same key.

Source: examples/slack_approval/. Sample code — outside the wheel, outside the public API, no compatibility promise beyond its own tests. Copy it into your own service and pin the beam-agents version you copied from.

What the surface does

Module	Role
surface.py	the consume → post → decide → publish loop; no Beam, no slack-sdk
slack.py	the SlackGateway seam: protocol, FakeSlackGateway, SocketModeGateway
blocks.py	Block Kit rendering and the button action value
config.py	URIs, channel, tokens, sweep interval — validated eagerly, import-free
agent.py	the FakeLLM demo agent and its pipeline wiring (the one module importing Beam)
__main__.py	python -m examples.slack_approval — real transports, graceful shutdown

Consume → post. Intents arrive through the effector's IntentSource seam (kafka://, pubsub://, or in-memory), so the example inherits per-key order, explicit commits, and the offline fakes instead of re-implementing them. Each live kind == APPROVAL intent becomes exactly one Block Kit message with approve/deny buttons; anything else is skipped and committed, so pointing the surface at the wrong topic is inert rather than destructive. The delivery is committed strictly after the post succeeds — a crash before posting re-delivers rather than loses.

Decide → publish. A verdict becomes one AgentEnvelope carrying Approval(intent_id, approved, approver, decided_at_ms), serialized with SerializeToString(deterministic=True) and published under the raw entity_key — the same keying WriteIntents and the effector's result sink use, so it lands on the suspended key's partition. approver is the Slack user id; decided_at_ms is the interaction's own timestamp, so the envelope is a function of the click and not of when the surface got around to it. Publish first, edit the message after: the envelope is the effect, the edit is cosmetic.

The pipeline stays the arbiter. The surface does not try to enforce at-most-one verdict globally. Racing clicks may publish two envelopes; the runtime's resume admission takes the first against the live continuation and orphans the rest to .errors. All the surface does is stop itself from re-publishing and answer later clicks as already decided.

TTL: expired means not actionable

Expiry uses the runtime's own guard, hitl.intent_expired, against an injectable clock — the same function the effector calls, with the same reading that a non-positive expires_at_ms is expired, never unbounded. It is applied at three points:

1. At consume time — an already-expired intent is posted as a non-interactive "expired before it could be surfaced" notice and committed. Posting rather than dropping keeps the channel an honest log of what was requested.
2. While pending — a periodic sweep (default 30 s) edits any message whose expiry passed: buttons removed, expired status shown. Cosmetic only; the layer-1 HITL timer is what actually resolves the suspension (deny by default).
3. At decision time — every interaction is re-checked against the expires_at_ms carried in the button's action value before an envelope is built. This is the load-bearing check, and it deliberately does not read in-process state: a click on a message posted by a previous process is still refused correctly.

Even a surface that got all three wrong is backstopped — _resume refuses a past-deadline approval. The example demonstrates cooperating with fail-closed layers, not being the only one.

Slack app setup (Socket Mode)

Socket Mode, not an HTTP Events endpoint: it needs no public HTTPS URL, so the demo runs on a laptop against docker compose, and there is no request-signature code in the example to copy wrong (see the trade-off below).

Create an app from this manifest, install it to your workspace, and invite the bot to the channel:

display_information:
  name: beam-agents approvals
features:
  bot_user:
    display_name: beam-agents
settings:
  interactivity:
    is_enabled: true
  socket_mode_enabled: true
oauth_config:
  scopes:
    bot:
      - chat:write

Then export the two tokens:

export SLACK_BOT_TOKEN=xoxb-...   # bot token, chat:write
export SLACK_APP_TOKEN=xapp-...   # app-level token, connections:write (Socket Mode)

Interactions arrive over the app-token-authenticated WebSocket. Slack requires each envelope to be acknowledged by echoing its envelope_id "so that Slack knows whether to retry", so SocketModeGateway acks first, before any chat.update; a redelivered interaction surfaces as a duplicate decision, which the surface collapses (first verdict wins).

If you deploy the HTTP alternative instead

Not implemented here, but stated honestly, because a production surface may need it. An HTTP interactivity endpoint must additionally:

• Verify the request signature. Slack signs each request with your app's signing secret: an HMAC-SHA256 over a versioned concatenation of the request timestamp and the raw body, compared in constant time against the X-Slack-Signature header, with a freshness window on X-Slack-Request- Timestamp to bound replay. Check the current Slack documentation for the exact version prefix, header names, and recommended window — do not copy these details from memory, including this paragraph's.
• Acknowledge within Slack's deadline, before doing the publish, and handle Slack's retries of anything it considers unacked (the same duplicate-decision tolerance the surface already has).
• Terminate TLS on a public endpoint and keep the raw body around for signature verification (frameworks that parse the body first make this subtly wrong).

Running the compose demo

make compose-up   # Redpanda on localhost:19092

python -m examples.slack_approval \
  --intents-from  kafka://localhost:19092/approval-requests \
  --approvals-to  kafka://localhost:19092/approvals \
  --slack-channel '#approvals'

Point --intents-from at the effector's --approvals-to channel (the primary wiring). For a minimal demo you can point it directly at the outbox and run no effector at all — sound because the demo agent stages only approval intents, and the surface's kind filter makes any stray intent inert.

The demo agent (agent.py) guards a pretend refund: it requests approval with a 10-minute TTL and suspends; approved it emits refund-issued, denied — or timed out — it emits refund-declined and the guarded action never happens. FakeLLM is the model, so no provider credentials are involved anywhere.

slack-sdk is only needed to run against a real workspace:

uv pip install slack-sdk

The offline tests never need it: SocketModeGateway imports it lazily in its constructor, exactly like the effector's transport adapters.

The surface, in full

ApprovalSurface is the whole out-of-pipeline loop: consume an approval intent, post it, take the verdict, publish the envelope back. This is the module itself, rendered from source — the code below is the code the tests below execute.

examples/slack_approval/surface.py

"""The approval surface: consume -> post -> decide -> publish.

One class, four obligations:

- **Consume -> post.** Approval-kind intents arrive through the effector's
  `IntentSource` seam; each live one becomes exactly one interactive Slack
  message, and the delivery is committed strictly *after* the post succeeds, so
  a crash re-delivers rather than loses. Non-approval kinds (including
  `TOOL_KIND_UNSPECIFIED`) are skipped and committed — pointing the surface at
  the wrong topic is inert, never destructive. Duplicate deliveries (the
  effector publishes approval notifications before marking them terminal) are
  collapsed by `intent_id` within the process; across a restart a redelivered
  intent may post again, which is harmless — both messages carry the same
  `intent_id` and the pipeline admits at most one verdict.

- **Decide -> publish.** A verdict becomes one deterministically serialized
  `AgentEnvelope.Approval` published under the raw `entity_key` through the
  `MessageSink` seam — the same keying `WriteIntents` and the effector's result
  sink use, so the envelope lands on the suspended key's partition. Publish
  first, edit after: the envelope is the effect, the edit is cosmetic. The
  surface never enforces at-most-one-verdict globally — the pipeline's resume
  admission is the arbiter; it only stops *itself* from re-publishing, and
  answers later clicks as already decided.

- **TTL, fail closed, at three points** (all via `hitl.intent_expired`, the
  runtime's own guard, against an injectable clock): already-expired intents
  are surfaced as a non-interactive notice; a periodic sweep edits pending
  messages whose expiry passed; and every decision is re-checked against the
  `expires_at_ms` carried in its action value — independent of in-process
  state, so it holds for clicks on messages posted before a restart. The
  first two are UX; the third is the gate. Even a surface that got all three
  wrong is backstopped by the pipeline: `_resume` refuses late approvals.

- **Bounded memory.** The posted/decided/pending maps are evicted by the sweep
  once entries are long past expiry, and hard-capped oldest-first.

Imports no Beam and no slack-sdk: this is an out-of-pipeline service, like the
effector it borrows its transport seams from.
"""

from __future__ import annotations

import asyncio
import logging
import time
from collections.abc import Callable
from dataclasses import dataclass

from beam_agents._protos import AgentEnvelope, ToolIntent
from beam_agents.effector.sinks import MessageSink
from beam_agents.effector.sources import DeliveredIntent, IntentSource
from beam_agents.hitl import intent_expired

from .blocks import approval_message, decided_edit, expired_edit, expired_notice
from .slack import Decision, MessageRef, SlackGateway

_LOG = logging.getLogger(__name__)

# How often the sweep pass runs; well under the demo's 10-minute TTL.
DEFAULT_SWEEP_INTERVAL_MS = 30_000

# Hard cap on tracked intent_ids (posted/decided/pending each); beyond it the
# oldest entries are evicted first. Restart semantics already tolerate an empty
# map, so eviction costs at worst a duplicate post or a re-answered click.
MAX_TRACKED_INTENTS = 4096

# How long a tracked entry outlives its expiry before the sweep evicts it:
# long enough to still collapse the channel's redelivery window.
_TRACKED_RETENTION_MS = 3_600_000


def _now_ms() -> int:
    return int(time.time() * 1000)


@dataclass(frozen=True)
class _Pending:
    """A posted, undecided approval: what the sweep needs to expire it."""

    message: MessageRef
    expires_at_ms: int


class ApprovalSurface:
    """The consume/decide loop over an `IntentSource`, a `SlackGateway`, and a
    `MessageSink`. Construct with in-memory fakes for tests and single-process
    demos, or with the Kafka/Pub/Sub adapters for a deployment.
    """

    def __init__(
        self,
        *,
        source: IntentSource,
        sink: MessageSink,
        gateway: SlackGateway,
        channel: str,
        sweep_interval_ms: int = DEFAULT_SWEEP_INTERVAL_MS,
        time_fn: Callable[[], int] = _now_ms,
    ) -> None:
        if not channel:
            raise ValueError("ApprovalSurface.channel must be a non-empty Slack channel")
        if sweep_interval_ms <= 0:
            raise ValueError(
                f"ApprovalSurface.sweep_interval_ms must be positive; got {sweep_interval_ms!r}"
            )
        self._source = source
        self._sink = sink
        self._gateway = gateway
        self._channel = channel
        self._sweep_interval_ms = sweep_interval_ms
        self._time_fn = time_fn
        # intent_id -> expires_at_ms for everything ever posted (interactive or
        # notice) this process lifetime; the within-process double-post guard.
        self._posted: dict[str, int] = {}
        # intent_id -> _Pending for posted, undecided, unexpired messages.
        self._pending: dict[str, _Pending] = {}
        # intent_id -> expires_at_ms for published verdicts; the re-publish guard.
        self._decided: dict[str, int] = {}
        self._stopping = asyncio.Event()

    # -- consume -> post -------------------------------------------------------

    async def consume(self) -> None:
        """Start the source and process deliveries until it is exhausted.

        A failed post propagates with the delivery uncommitted — crash-stop is
        the crash-safety story: the transport re-delivers to the next instance.
        """
        await self._source.start()
        await self._consume_started()

    async def _consume_started(self) -> None:
        async for delivered in self._source:
            await self._handle_delivery(delivered)

    async def _handle_delivery(self, delivered: DeliveredIntent) -> None:
        intent = delivered.intent
        if intent.kind != ToolIntent.APPROVAL:
            # Not ours (TOOL_KIND_UNSPECIFIED included): skip and commit. A
            # TOOL intent can never become a button that fabricates a verdict.
            await self._source.commit(delivered)
            return
        if intent.intent_id in self._posted:
            # The effector notifies before marking terminal, so duplicates are
            # expected; one interactive message per intent_id per process.
            await self._source.commit(delivered)
            return
        if intent_expired(intent, self._time_fn()):
            text, blocks = expired_notice(intent)
            await self._gateway.post(self._channel, text=text, blocks=blocks)
            self._track_posted(intent)
        else:
            text, blocks = approval_message(intent)
            ref = await self._gateway.post(self._channel, text=text, blocks=blocks)
            self._track_posted(intent)
            self._pending[intent.intent_id] = _Pending(
                message=ref, expires_at_ms=intent.expires_at_ms
            )
        # Commit strictly after the post: a crash between them re-delivers and
        # (worst case, across a restart) re-posts — never loses the request.
        await self._source.commit(delivered)

    def _track_posted(self, intent: ToolIntent) -> None:
        self._posted[intent.intent_id] = intent.expires_at_ms
        _evict_oldest(self._posted)

    # -- decide -> publish -----------------------------------------------------

    async def process_decisions(self) -> None:
        """Handle gateway decisions until its stream ends (gateway closed)."""
        async for decision in self._gateway.decisions():
            await self.handle_decision(decision)

    async def handle_decision(self, decision: Decision) -> None:
        """One interaction: expiry-gate, publish, edit — in that order."""
        if decision.intent_id in self._decided:
            # First verdict won and was published; nothing further leaves the
            # surface for this intent. The pipeline would orphan a duplicate
            # anyway — this just answers the human instead of racing them.
            await self._gateway.answer(
                decision, f"Already decided: intent {decision.intent_id} has a published verdict."
            )
            return
        now_ms = self._time_fn()
        # THE fail-closed gate (design D6.3): checked against the expiry the
        # button value carries, not against `_pending`, so it survives sweeps
        # that have not run yet and restarts that emptied the maps.
        if intent_expired(ToolIntent(expires_at_ms=decision.expires_at_ms), now_ms):
            self._pending.pop(decision.intent_id, None)
            text, blocks = expired_edit(decision.intent_id)
            await self._gateway.update(decision.message, text=text, blocks=blocks)
            return
        entity_key = bytes.fromhex(decision.entity_key_hex)
        envelope = AgentEnvelope(entity_key=entity_key, event_time_ms=decision.decided_at_ms)
        envelope.approval.intent_id = decision.intent_id
        envelope.approval.approved = decision.approved
        envelope.approval.approver = decision.approver
        envelope.approval.decided_at_ms = decision.decided_at_ms
        # Deterministic bytes under the raw entity_key: lands on the suspended
        # key's partition/ordering key, byte-stable for replay comparison.
        await self._sink.publish(entity_key, envelope.SerializeToString(deterministic=True))
        self._decided[decision.intent_id] = decision.expires_at_ms
        _evict_oldest(self._decided)
        self._pending.pop(decision.intent_id, None)
        # Publish, then edit: the envelope is the effect, the edit is cosmetic.
        # A crash between them leaves a decided-but-live-looking message whose
        # next click is refused above.
        text, blocks = decided_edit(
            decision.intent_id, approved=decision.approved, approver=decision.approver
        )
        await self._gateway.update(decision.message, text=text, blocks=blocks)

    # -- the sweep -------------------------------------------------------------

    async def sweep_once(self) -> None:
        """Edit pending messages whose expiry passed; evict long-dead tracking.

        UX only — the layer-1 HITL timer resolves the suspension and the
        decision-time gate refuses late clicks regardless; the sweep just stops
        the UI from soliciting clicks that can no longer matter.
        """
        now_ms = self._time_fn()
        for intent_id, pending in list(self._pending.items()):
            if not intent_expired(ToolIntent(expires_at_ms=pending.expires_at_ms), now_ms):
                continue
            del self._pending[intent_id]
            text, blocks = expired_edit(intent_id)
            await self._gateway.update(pending.message, text=text, blocks=blocks)
        for tracked in (self._posted, self._decided):
            for intent_id, expires_at_ms in list(tracked.items()):
                if expires_at_ms + _TRACKED_RETENTION_MS <= now_ms:
                    del tracked[intent_id]

    async def _sweep_loop(self) -> None:
        while True:
            await asyncio.sleep(self._sweep_interval_ms / 1000)
            await self.sweep_once()

    # -- lifecycle -------------------------------------------------------------

    async def run(self) -> None:
        """Run consume, decisions, and the sweep until `stop()` or a failure.

        Shutdown order: stop consuming (workers cancelled — an in-flight post
        finishes or its delivery stays uncommitted), then close the gateway,
        the sink, and the source. A worker failure propagates after cleanup:
        crash-stop, so the transport's redelivery does the recovery.
        """
        await self._source.start()
        workers: set[asyncio.Task[None]] = {
            asyncio.create_task(self._consume_started(), name="consume"),
            asyncio.create_task(self.process_decisions(), name="decisions"),
            asyncio.create_task(self._sweep_loop(), name="sweep"),
        }
        stop_waiter = asyncio.create_task(self._stopping.wait(), name="stop")
        failure: BaseException | None = None
        try:
            while workers:
                done, _ = await asyncio.wait(
                    workers | {stop_waiter}, return_when=asyncio.FIRST_COMPLETED
                )
                if stop_waiter in done:
                    break
                for task in done & workers:
                    workers.discard(task)
                    if task.exception() is not None:
                        failure = task.exception()
                        break
                if failure is not None:
                    break
        finally:
            stop_waiter.cancel()
            for task in workers:
                task.cancel()
            await asyncio.gather(*workers, stop_waiter, return_exceptions=True)
            await self._gateway.close()
            await self._sink.close()
            await self._source.close()
        if failure is not None:
            raise failure

    def stop(self) -> None:
        """Ask `run()` to shut down; safe to call from a signal handler."""
        self._stopping.set()


def _evict_oldest(tracked: dict[str, int], cap: int = MAX_TRACKED_INTENTS) -> None:
    """Drop insertion-oldest entries until `tracked` fits the cap."""
    while len(tracked) > cap:
        oldest = next(iter(tracked))
        _LOG.warning("tracking cap reached; forgetting intent %s", oldest)
        del tracked[oldest]

What the tests hold this page to

tests/examples/test_slack_approval.py imports the same modules this page describes, so a change that invalidates the documented demo fails the tests.

• Offline leg (make test-unit; no docker, no workspace, no slack-sdk): the FakeLLM demo activation stages a real intent under TestPipeline, and the surface consumes it from InMemoryIntentSource, posts to FakeSlackGateway, takes a scripted click, and publishes into InMemoryMessageSink. Asserted: the Block Kit action values, raw-entity_key keying, byte-exact envelope serialization, commit-after-post, skip-and-commit for non-approval kinds, within-process duplicate collapse, all three TTL points (including the post-restart click), and — fed back through a TestStream-scripted pipeline — that the published envelope resumes the suspended activation on both the approve and the deny path.
• Compose-Kafka leg (tests/examples/test_slack_approval_kafka.py, -m integration): the same loop over real Redpanda topics, with the envelope read back off the wire and re-injected byte-for-byte.

Residuals (stated, not engineered away)

• A duplicate message after a surface restart. The channel is at-least-once by design (the effector publishes the approval notification before marking it terminal). Within one process the surface collapses redeliveries by intent_id; across a restart its map is empty, so a redelivered intent may post a second message. Harmless: both carry the same intent_id, whichever is answered first wins, and the pipeline admits at most one verdict. A durable posted-set would be the effector's dedup machinery again — more than this example is for.
• The pending map is lost on restart. Only the sweep's cosmetic edits are affected; the decision-time expiry check rides on the button value, so fail-closed behavior survives.
• No approval audit store. The Slack channel history is the demo's audit trail. A production surface would add its own.
• No exactly-once posting to Slack, and no at-most-one-verdict enforcement at the surface — see "the pipeline stays the arbiter" above.