Human-in-the-loop approval: pluggable approver (decision model first, human fallback) - design discussion for #1067

[Ashfaqbs]

Follow-up discussion for #1067, as requested there.

Problem

A common agentic pattern is pausing a workflow for a human to approve a consequential step (send this email, execute this trade, merge this change) before continuing. Flink Agents already has the durable, checkpointed, keyed state that makes this pattern natural to support well, but the runtime questions raised on #1067 need answering before any implementation:

• An unfinished ActionTask keeps the key in flight, and later events for the same key are buffered. An approval delivered as an ordinary keyed input event could be queued behind the Action it needs to resume.
• A long-lived pending key may affect watermark progression and end-of-input handling.
• Checkpoints persist operator-managed task state, but not an in-memory Java stack or Python coroutine. Recovery needs an explicit persisted approval state machine and deterministic re-entry.
• [Tech Debt] Align Java and Python recovery model #1016 aligns part of the Java/Python recovery model, but doesn't yet define durable Action suspension and external resumption.

A narrower first version: pluggable approver

Rather than treating every proposal as needing the full durable-suspend-resume path, treat the approver as pluggable, with a decision model as the first tier and a human as the fallback (credit to @dangzitou for this framing).

Use case. In cross-border e-commerce, for example, an agent proposes actions such as raising an ad bid or pausing a listing. Most proposals are routine and could be cleared automatically; only ambiguous or high-impact ones need a person. Sending every proposal to a human makes approval the bottleneck, and every one of them would take the hard suspend/resume path.

What already works today, with no new runtime semantics, now that DecisionModel resources have landed (#1243 / #1264):

1. The proposing Action emits a DecisionRequestEvent with a PredicateQuestion (e.g. "Is this change within policy?") and application-supplied criteria.
2. The DecisionResponseEvent handler applies application-owned thresholds:
   • probability_true >= T_approve -> continue to the next step (e.g. emit the event that triggers the tool call)
   • probability_true <= T_reject -> reject and emit an output
   • in between, a RefusalAnswer, or a failed response -> escalate to a human

Because the call goes through the built-in Decision Action, its outcome is recorded via durable execution when an Action State Store is configured, so recovery replays the verdict instead of asking the model again.

What this means for scope:

• Only the escalation branch needs durable suspension. The questions above still need answers, but they'd apply to a much smaller share of records, which may make a narrower first version acceptable.
• If the pending-approval state records who resolved it (a decision model with its probabilities, or a human with an identity), the audit trail stays uniform across both paths.
• Thresholds and the escalation policy should stay in application code, consistent with the DecisionModel design, where the business Action "chooses thresholds, and decides what to do with the answers."

Open questions for maintainers

1. Is a built-in await_approval-style Action type the right shape, or should this stay a documented pattern built from existing primitives (Decision Action + a resource/tool call for the human-escalation leg)?
2. For the escalation leg specifically, how should the external "approve" signal be delivered - a resource/tool call, an API, or a keyed Kafka topic tied to the pending approval id?
3. Does narrowing scope to "only the escalation branch needs durable suspension" change the answers to the in-flight-key-buffering / watermark / recovery questions raised on [Feature] Human-in-the-loop approval as a first-class, checkpointed Action primitive #1067, or do they need full answers regardless of how small the escalation share is?

Related: #1067, #1016, #1243, #1264

[renezander030]

The narrower version makes one recovery case more common: if the job fails after an approved call has reached the ad platform but before its result is checkpointed, recovery replays the decision verdict and can send the same bid change a second time. I'd carry the pending-approval id into that call as its idempotency key and record the outcome as succeeded, failed or unknown next to the verdict, so recovery can tell an effect that already happened from one that never went out.

I keep the consumed-action check in the same transaction as the dispatch record here: https://gist.github.com/renezander030/ad81c7a805a09a844983f881e2c487e5

[Ashfaqbs]

Good catch — that's exactly the race I hadn't pinned down: a decision replayed on recovery, but the downstream effect's own completion was never
checkpointed. Carrying the pending-approval id as the idempotency key into that call is the right fix, and recording succeeded/failed/unknown next to
the verdict matters a lot — without "unknown" you can't tell a timeout from a dispatch that genuinely never went out, which is the difference that
matters for something like an ad-spend or trade action you can't just blindly retry.

Keeping the consumed-action check and the dispatch record in the same transaction (thanks for the gist) closes the window where the effect fires but we
forget we dispatched it. I'd fold this into question 1 above — it's really the audit-trail/outcome-recording half of that question. Mind if I link your
comment from the issue body as the answer to that part?