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

["Ashfaqbs (via GitHub)" <gi...@apache.org>]

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

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.
- #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 #1067, or do they need full answers regardless of how small the escalation share is?

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


GitHub link: https://github.com/apache/flink-agents/discussions/1304

----
This is an automatically sent email for issues@flink.apache.org.
To unsubscribe, please send an email to: issues-unsubscribe@flink.apache.org

Re: [D] Human-in-the-loop approval: pluggable approver (decision model first, human fallback) - design discussion for #1067 [flink-agents]

["Ashfaqbs (via GitHub) via issues" <is...@flink.apache.org>]

GitHub user Ashfaqbs added a comment to the discussion: Human-in-the-loop approval: pluggable approver (decision model first, human fallback) - design discussion for #1067

That `InputEvent` variant for approval responses matches what I'd want here, with one addition on the "avoid being buffered behind the task it needs to resume" requirement: the event needs to skip the operator's normal per-key action-queue entirely, not just be prioritized within it. If it goes through the same enqueue path as an ordinary keyed input, it still has to wait for its turn behind whatever's ahead of it for that key, which includes the task it's meant to unblock. The resume event needs its own short-circuit at `processElement` (or wherever keys get dispatched to the queue) that looks up the suspended task by its pending-approval id and resumes it directly, bypassing the FIFO queue for that key rather than jumping it to the front.

On timeout: I'd model it as a second event type racing the approval response rather than a special case inside it - a `ApprovalTimeoutEvent` carrying the same pending-approval id, scheduled via a processing-time timer when the task suspends. Whichever arrives first resumes the task and the other becomes a no-op (the suspended-task lookup by id returns nothing, since it's already gone). That keeps "what happens on timeout" as application-visible behavior (the resumed task sees "timed out" vs "approved" as the resume payload) instead of buried runtime policy, and it reuses the exact same bypass-the-queue resume path rather than needing a second mechanism.

On duplicate/late responses: this is the same problem as the idempotency-key point above, just one level up the stack. Once the suspended task is resumed and removed from the pending-approval index, a second response (duplicate delivery, or a human approving after a timeout already resumed it as rejected) finds nothing to resume by that id and should be dropped with a logged warning rather than erroring - silently accepting it twice is exactly the double-dispatch risk from the gist. If the audit trail needs to show a late response was discarded rather than just vanishing, that's a case for writing it to the pending-approval's `alert_data`-equivalent record *after* it's no longer pending, as a "late, ignored" entry rather than reopening the resumed state.

Suspension surviving recovery: since the suspension state (pending-approval id, the key, what to resume with) has to be durable action state anyway for the resume-by-id lookup to work after a restart, that's the same action-state store this thread's already assuming exists per #1067 - no separate persistence mechanism needed, just confirming recovery replays "still pending" rather than re-triggering the original request.

GitHub link: https://github.com/apache/flink-agents/discussions/1304#discussioncomment-19379425

----
This is an automatically sent email for issues@flink.apache.org.
To unsubscribe, please send an email to: issues-unsubscribe@flink.apache.org

Re: [D] Human-in-the-loop approval: pluggable approver (decision model first, human fallback) - design discussion for #1067 [flink-agents]

["wenjin272 (via GitHub) via issues" <is...@flink.apache.org>]

GitHub user wenjin272 added a comment to the discussion: Human-in-the-loop approval: pluggable approver (decision model first, human fallback) - design discussion for #1067

Thanks for the discussion. I don't have a concrete design yet, but I see using a Decision Model for automatic approval as one possible approval policy. I agree that the policy should be pluggable.

To me, the main challenge is how to suspend an ongoing task and resume it when human feedback arrives. We should also support a timeout mechanism so the application can decide what to do when an approval request remains unanswered for too long.

One initial idea is to introduce a new `InputEvent` type representing an approval response for a previous Agent Run. Instead of following the normal input-processing path, this event would locate and resume the suspended task associated with that key. It would need to identify the specific run and approval request, rather than relying on the key alone, and avoid being buffered behind the task it needs to resume.

This is only a starting point. We would still need to define how suspension survives recovery, how approval responses and timeouts interact, and how duplicate or late responses are handled. The point raised above about external side effects is also relevant: resuming an approved task does not by itself prevent an external operation from being executed twice.

GitHub link: https://github.com/apache/flink-agents/discussions/1304#discussioncomment-19345268

----
This is an automatically sent email for issues@flink.apache.org.
To unsubscribe, please send an email to: issues-unsubscribe@flink.apache.org

Re: [D] Human-in-the-loop approval: pluggable approver (decision model first, human fallback) - design discussion for #1067 [flink-agents]

["renezander030 (via GitHub)" <gi...@apache.org>]

GitHub user renezander030 added a comment to the discussion: Human-in-the-loop approval: pluggable approver (decision model first, human fallback) - design discussion for #1067

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

GitHub link: https://github.com/apache/flink-agents/discussions/1304#discussioncomment-19176442

----
This is an automatically sent email for issues@flink.apache.org.
To unsubscribe, please send an email to: issues-unsubscribe@flink.apache.org

Re: [D] Human-in-the-loop approval: pluggable approver (decision model first, human fallback) - design discussion for #1067 [flink-agents]

["Ashfaqbs (via GitHub)" <gi...@apache.org>]

GitHub user Ashfaqbs added a comment to the discussion: Human-in-the-loop approval: pluggable approver (decision model first, human fallback) - design discussion for #1067


   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?

GitHub link: https://github.com/apache/flink-agents/discussions/1304#discussioncomment-19200423

----
This is an automatically sent email for issues@flink.apache.org.
To unsubscribe, please send an email to: issues-unsubscribe@flink.apache.org