Approvals and timeout fallback

Asking a person is easy. Deciding what happens when nobody answers is the part that has to be designed, and it is the part this example is about.

Two keys request the same approval. One gets an answer at 1.2 seconds. The other is never answered at all, its timer fires, and the policy converts the silence into a deterministic denial. Both outcomes are asserted in one bounded run.

Two keys, two outcomes

acct-answered — an Approval matching the staged intent id arrives. The activation resumes and emits b"approved".
acct-silent — nothing arrives. The HITL_TIMER fires, the policy returns Deny, and the denial bytes are emitted instead.

The second case is the interesting one. An approval that never arrives cannot be allowed to leave a key suspended forever, and it cannot be allowed to take effect later either.

Asking: request_approval, then Suspend

async def large_transfer(ctx: ActivationContext) -> Complete | Suspend:
    """Hold a large transfer until a human approves it."""
    if not ctx.is_resume:
        ctx.request_approval('{"amount": 250000}', ttl_ms=APPROVAL_TTL_MS)
        return Suspend(
            snapshot=b"awaiting-approval",
            adapter="example",
            timeout_ms=SUSPENSION_TIMEOUT_MS,
        )

    approval = ctx.resume_approval
    assert approval is not None
    return Complete(output=b"approved" if approval.approved else b"rejected")

website/examples/human_in_the_loop.py (region: agent) — executed by the repository’s offline test tier.

The agent is invoked twice for one logical unit of work, and ctx.is_resume is how it tells the two invocations apart.

ctx.request_approval stages an approval intent and returns its deterministic id. Suspend persists a continuation in keyed state and arms the real-time deadline; timeout_ms overrides HitlPolicy.timeout_ms for this suspension. When the first invocation returns, nothing has reached the outside world — the intent is a request, and the activation is simply over. The worker moves on to other keys.

On resume, ctx.resume_approval carries the answer. The snapshot the agent persisted comes back as ctx.snapshot, which is where a real agent would keep whatever it needs to finish the job.

Deciding what silence means

def deny_on_timeout(fallback: FallbackContext) -> Route:
    """Route an unanswered approval to a deterministic denial.

    Pure and synchronous. It reads only the `FallbackContext` it is handed —
    which carries the suspended `seq`, the persisted snapshot, the elapsed
    deadline, the timer's fire time, and the intent ids nothing answered — so a
    retried timer bundle reaches the same decision.
    """
    return Deny(output=b"denied:no-approval:" + str(fallback.seq).encode())

website/examples/human_in_the_loop.py (region: policy) — executed by the repository’s offline test tier.

on_timeout returns one of three routes:

Deny(output=...) — emit deterministic bytes on .output and end the suspension. This example's choice.
Drop(reason=...) — emit nothing, and record the timeout on .errors as hitl_timeout.
Escalate(tool_name=..., timeout_ms=...) — stage a fresh approval on another channel and extend the deadline, bounded by HitlPolicy.max_escalations (default 0). An unbounded escalate loop would be a fail-open hole; the whole point of the timer is that the wait ends.

The function must be pure, synchronous, and picklable. That is a correctness requirement rather than a style rule. It runs inside a timer callback, and a timer callback re-executes when its bundle is retried — a policy that read a clock, called the model, or drew unseeded randomness would reach a different decision the second time and the retry would diverge from the original.

So everything the policy could need is handed to it on the FallbackContext: the suspended seq, the persisted snapshot, the deadline_ms that elapsed, the timer's fired_at_ms, and the pending_intent_ids nothing ever answered. Note what the example does with fallback.seq — the denial bytes are derived from the context, so the output is a function of the suspension rather than of when the timer happened to fire.

Picklable means a module-level function, never a lambda or a closure: the DoFn holds the policy and serializes it for the runner.

Attaching the policy

# `approval_channel` names where the effector routes the request — a queue,
# a pager — not a registered tool, so nothing is resolved or executed for
# it in-pipeline. `intent_ttl_ms` is the default expiry stamped onto staged
# intents, and it is layer 2 of the fail-closed rule: past `expires_at_ms`
# the effector refuses the intent rather than acting on it.
policy = HitlPolicy(
    timeout_ms=SUSPENSION_TIMEOUT_MS,
    intent_ttl_ms=APPROVAL_TTL_MS,
    approval_channel="approval",
    on_timeout=deny_on_timeout,
)

website/examples/human_in_the_loop.py (region: policy-config) — executed by the repository’s offline test tier.

HitlPolicy validates itself on construction, so a non-positive timeout_ms or an empty approval_channel raises at this line rather than on a worker.

Shaping an approval

def _approval(
    key: bytes, intent_id: str, *, approved: bool, t_ms: int
) -> TimestampedValue[AgentEnvelope]:
    env = AgentEnvelope(entity_key=key, event_time_ms=t_ms)
    # `Approval` carries no entity_key of its own: the envelope's key is what
    # routes it, and the intent id is what matches it to a continuation.
    env.approval.intent_id = intent_id
    env.approval.approved = approved
    env.approval.approver = "ops@example.invalid"
    return TimestampedValue(env, t_ms / 1000)

website/examples/human_in_the_loop.py (region: approval) — executed by the repository’s offline test tier.

An approval re-enters the pipeline as an ordinary AgentEnvelope on the same key as the event that asked for it. It is not a callback and not a side channel — the architecture diagram in architecture shows the approvals topic flattened in alongside the events topic, and the runner cannot tell the two apart.

Two fields do the routing. The envelope's entity_key selects the key, and therefore the suspended continuation. The approval's intent_id selects which pending intent on that key is being answered. The Approval message carries no entity_key of its own, which is why the example sets it on the envelope.

The id the test uses is computed before the pipeline runs:

# Two keys: one gets an answer, one never does. The answered key's approval
# can be addressed before the pipeline exists, because the intent id is a
# pure function of (key, seq, step_index): first activation of the key is
# seq 0, and the approval it stages is step 0.
answered = intent_id_for(b"acct-answered", 0, 0)

website/examples/human_in_the_loop.py (region: intent-id) — executed by the repository’s offline test tier.

Because the id is uuid5 over exactly those coordinates it can be named in advance — the same property intents and resume leans on, and the basis of the effectively-once argument.

Scripting both clocks

stream = (
    TestStream()
    .advance_watermark_to(0)
    .add_elements([_event(b"acct-answered", b"transfer", 1_000)])
    .add_elements([_event(b"acct-silent", b"transfer", 1_000)])
    .add_elements([_approval(b"acct-answered", answered, approved=True, t_ms=1_200)])
    # Push processing time past the suspension timeout so the silent key's
    # HITL timer fires. Scripted, never a sleep: the test controls both
    # clocks, so the outcome is deterministic.
    .advance_processing_time(60)
    .advance_watermark_to_infinity()
)

website/examples/human_in_the_loop.py (region: stream) — executed by the repository’s offline test tier.

The suspension timeout lives in processing time, so this pipeline needs a processing clock to move. TestStream.advance_processing_time moves it, deliberately, by a scripted amount — the example never sleeps.

That matters for more than speed. A sleep would make the test's outcome depend on how loaded the machine is, which is the definition of a flaky gate. Advancing both the watermark and the processing clock by hand makes the timer fire at a point the test chose, so the assertion below is the only result the pipeline can produce. It is the same technique the repository's own timer tests use.

The assertion

outputs = keyed | "Agent" >> RunAgent(
    large_transfer,
    config=AgentConfig(
        provider_factory=make_provider,
        hitl_policy=policy,
        ttl_ms=1_000_000_000,
    ),
)

# One assertion covers both keys: the answered one resumes and
# completes, the silent one is routed by the policy. Neither outcome is
# a timing accident — both are the only value the pipeline can produce.
assert_that(
    outputs.output,
    equal_to([b"approved", b"denied:no-approval:0"]),
    label="approved-and-denied",
)

website/examples/human_in_the_loop.py (region: wiring) — executed by the repository’s offline test tier.

ttl_ms is set very high on purpose: working-memory TTL collection would otherwise be free to wipe the suspended continuation during a long scripted run, and the example is not trying to demonstrate that.

Failing closed at both layers

The timer above is layer 1, inside the pipeline. There is a second, independent guard outside it: the effector refuses any intent whose expires_at_ms has passed, treating a non-positive expiry as expired rather than as unbounded.

Both layers are needed. Layer 1 alone would end the wait but still leave a late approval able to cause an effect afterwards; layer 2 alone would leave the key suspended forever. Together they mean the runtime's decision to give up is final. Results and approvals that arrive with no live continuation are dropped to .errors as orphaned_result, with a detail naming why.

This is correctness invariant 6, and tests/semantics/test_hitl_fail_closed.py is the gate that holds it.

Run it

uv run python website/examples/human_in_the_loop.py

The complete program

"""Human-in-the-loop: ask a person, and decide what happens when nobody answers.

`ctx.request_approval` stages an approval intent on the configured channel and
`Suspend` parks the activation. If the approval arrives, the agent resumes with
it. If it never arrives, the HITL timer fires and `HitlPolicy.on_timeout`
decides — `Deny` (emit deterministic bytes and end), `Drop` (emit nothing and
record the timeout on `.errors`), or `Escalate` (ask again, louder, bounded by
`max_escalations`).

Timeouts fail closed at both layers. This file shows layer 1, the in-pipeline
timer. Layer 2 is the effector refusing an intent past its `expires_at_ms`, so a
late approval cannot cause an effect after the runtime has already given up.

`on_timeout` must be pure, synchronous, and picklable — a module-level function,
never a lambda. That is a correctness requirement rather than a style rule: a
timer callback re-executes when its bundle is retried, and a fallback that read
a clock or called the model would make the retry diverge from the original.
Every value the policy could need is carried on the `FallbackContext`.

Run it:  python website/examples/human_in_the_loop.py
"""

from __future__ import annotations

import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions, StandardOptions
from apache_beam.testing.test_stream import TestStream
from apache_beam.testing.util import assert_that, equal_to
from apache_beam.transforms.window import TimestampedValue

from beam_agents import AgentConfig, Deny, FallbackContext, HitlPolicy, RunAgent
from beam_agents._protos import AgentEnvelope
from beam_agents.core.agent import Complete, Suspend, intent_id_for
from beam_agents.core.context import ActivationContext
from beam_agents.hitl import Route
from beam_agents.model.fake import FakeLLM, match_any, respond_with

APPROVAL_TTL_MS = 60_000
SUSPENSION_TIMEOUT_MS = 1_000


def make_provider() -> FakeLLM:
    return FakeLLM([(match_any(), respond_with(b"ok"))])


# region: policy
def deny_on_timeout(fallback: FallbackContext) -> Route:
    """Route an unanswered approval to a deterministic denial.

    Pure and synchronous. It reads only the `FallbackContext` it is handed —
    which carries the suspended `seq`, the persisted snapshot, the elapsed
    deadline, the timer's fire time, and the intent ids nothing answered — so a
    retried timer bundle reaches the same decision.
    """
    return Deny(output=b"denied:no-approval:" + str(fallback.seq).encode())


# endregion: policy


# region: agent
async def large_transfer(ctx: ActivationContext) -> Complete | Suspend:
    """Hold a large transfer until a human approves it."""
    if not ctx.is_resume:
        ctx.request_approval('{"amount": 250000}', ttl_ms=APPROVAL_TTL_MS)
        return Suspend(
            snapshot=b"awaiting-approval",
            adapter="example",
            timeout_ms=SUSPENSION_TIMEOUT_MS,
        )

    approval = ctx.resume_approval
    assert approval is not None
    return Complete(output=b"approved" if approval.approved else b"rejected")


# endregion: agent


def _event(key: bytes, payload: bytes, t_ms: int) -> TimestampedValue[AgentEnvelope]:
    env = AgentEnvelope(entity_key=key, event_time_ms=t_ms, external_event=payload)
    return TimestampedValue(env, t_ms / 1000)


# region: approval
def _approval(
    key: bytes, intent_id: str, *, approved: bool, t_ms: int
) -> TimestampedValue[AgentEnvelope]:
    env = AgentEnvelope(entity_key=key, event_time_ms=t_ms)
    # `Approval` carries no entity_key of its own: the envelope's key is what
    # routes it, and the intent id is what matches it to a continuation.
    env.approval.intent_id = intent_id
    env.approval.approved = approved
    env.approval.approver = "ops@example.invalid"
    return TimestampedValue(env, t_ms / 1000)


# endregion: approval


def main() -> None:
    # region: intent-id
    # Two keys: one gets an answer, one never does. The answered key's approval
    # can be addressed before the pipeline exists, because the intent id is a
    # pure function of (key, seq, step_index): first activation of the key is
    # seq 0, and the approval it stages is step 0.
    answered = intent_id_for(b"acct-answered", 0, 0)
    # endregion: intent-id

    # region: stream
    stream = (
        TestStream()
        .advance_watermark_to(0)
        .add_elements([_event(b"acct-answered", b"transfer", 1_000)])
        .add_elements([_event(b"acct-silent", b"transfer", 1_000)])
        .add_elements([_approval(b"acct-answered", answered, approved=True, t_ms=1_200)])
        # Push processing time past the suspension timeout so the silent key's
        # HITL timer fires. Scripted, never a sleep: the test controls both
        # clocks, so the outcome is deterministic.
        .advance_processing_time(60)
        .advance_watermark_to_infinity()
    )
    # endregion: stream

    options = PipelineOptions()
    options.view_as(StandardOptions).streaming = True

    # region: policy-config
    # `approval_channel` names where the effector routes the request — a queue,
    # a pager — not a registered tool, so nothing is resolved or executed for
    # it in-pipeline. `intent_ttl_ms` is the default expiry stamped onto staged
    # intents, and it is layer 2 of the fail-closed rule: past `expires_at_ms`
    # the effector refuses the intent rather than acting on it.
    policy = HitlPolicy(
        timeout_ms=SUSPENSION_TIMEOUT_MS,
        intent_ttl_ms=APPROVAL_TTL_MS,
        approval_channel="approval",
        on_timeout=deny_on_timeout,
    )
    # endregion: policy-config

    with beam.Pipeline(options=options) as pipeline:
        keyed = (
            pipeline
            | stream
            | "Key"
            >> beam.WithKeys(lambda e: e.entity_key).with_output_types(tuple[bytes, AgentEnvelope])
        )
        # region: wiring
        outputs = keyed | "Agent" >> RunAgent(
            large_transfer,
            config=AgentConfig(
                provider_factory=make_provider,
                hitl_policy=policy,
                ttl_ms=1_000_000_000,
            ),
        )

        # One assertion covers both keys: the answered one resumes and
        # completes, the silent one is routed by the policy. Neither outcome is
        # a timing accident — both are the only value the pipeline can produce.
        assert_that(
            outputs.output,
            equal_to([b"approved", b"denied:no-approval:0"]),
            label="approved-and-denied",
        )
        # endregion: wiring

    print("human_in_the_loop: ok")


if __name__ == "__main__":
    main()

website/examples/human_in_the_loop.py — executed by the repository’s offline test tier.

Human in the loop — the operational reference: channels, the route table, escalation bounds, and the orphan details.
Intents and resume — the same suspend/resume machinery, driven by an effector result instead of a person.
The effector — layer 2 of the fail-closed rule.
The errors output — where hitl_timeout and orphaned_result land.
HitlPolicy, Deny, Drop, Escalate, and FallbackContext in the API reference.