flow: human-in-the-loop pause/resume (durable workflow, stage A) - #4852
Merged
Conversation
Adds a waiting run state so a flow step can suspend for external input and resume durably — stage A of the durable-agentic-workflow design in #4816. - flow.Await(key, prompt) / flow.AwaitStep(...): a StepFunc that suspends the run. runFrom recognizes the signal, checkpoints the run with status "waiting" (recording what it awaits), and returns cleanly — a suspend is not a failure, and it is not retried or graded. - Flow.ResumeWith(ctx, runID, input): completes the awaited step with the injected input (which becomes that step's output state) and continues from the next step. - Flow.Waiting(ctx): lists suspended runs with their Await metadata. - ResumePending/Pending skip waiting runs — they need input, not a restart. Existing crash-resume (Resume) is unchanged. Additive: no signature or default-behavior changes. Await ergonomics (sentinel-return) are the default proposed in #4816; open to AwaitStep-kind instead if preferred. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01CmdEY7pYmV5zzwCjNJ4ykL
There was a problem hiding this comment.
Pull request overview
Adds human-in-the-loop (HITL) suspension to flow/ runs via a new durable waiting state, plus APIs to list and resume those suspended runs with injected external input. This extends the existing checkpoint/resume mechanism without changing existing flow step signatures.
Changes:
- Introduces
flow.Await(key, prompt)/flow.AwaitStep(...)to suspend a run and checkpoint it aswaitingwithRun.Awaitmetadata. - Adds
Flow.Waiting(ctx)to list suspended runs andFlow.ResumeWith(ctx, runID, input)to inject input and continue from the next step. - Updates OTel step spans so suspends are marked as OK (control flow), not errors, and updates
Pending/ResumePendingbehavior to skip waiting runs.
Reviewed changes
Copilot reviewed 4 out of 4 changed files in this pull request and generated 1 comment.
| File | Description |
|---|---|
| flow/steps.go | Adds waiting run status, await control signal + metadata, and waiting/resume APIs; updates engine execution to checkpoint and return cleanly on suspend. |
| flow/steps_test.go | Adds coverage for suspend/resume behavior and ensures waiting runs are excluded from Pending(). |
| flow/otel.go | Marks awaited suspends as codes.Ok rather than erroring the span. |
| CHANGELOG.md | Documents the new flow HITL pause/resume feature and APIs. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
Comment on lines
+516
to
+530
| run, ok, err := f.checkpoint.Load(ctx, runID) | ||
| if err != nil { | ||
| return err | ||
| } | ||
| if !ok { | ||
| return fmt.Errorf("run %s not found", runID) | ||
| } | ||
| if run.Status != "waiting" { | ||
| return fmt.Errorf("run %s is not waiting for input (status %q)", runID, run.Status) | ||
| } | ||
| steps := f.opts.Steps | ||
| i := stepIndex(steps, run.State.Stage) | ||
| if i < 0 { | ||
| return fmt.Errorf("run %s is waiting at unknown step %q", runID, run.State.Stage) | ||
| } |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Stage A of #4816 — the
waitingrun state + HITL pause/resume. This is the smallest, additive slice from the design proposal; opening it as concrete code to react to rather than auto-merging (the issue isneeds-human).What it adds
A flow step can now suspend a run pending external input and resume durably:
flow.Await(key, prompt)(and theflow.AwaitStep(name, key, prompt)convenience) — aStepFuncthat suspends the run.runFromrecognizes the signal, checkpoints the run with statuswaiting(recording what it awaits onRun.Await), and returns cleanly — a suspend is not a failure, and it is neither retried nor graded.Flow.ResumeWith(ctx, runID, input)— completes the awaited step with the injected input (which becomes that step's outputState) and continues from the next step.Flow.Waiting(ctx)— lists suspended runs with theirAwaitmetadata, so a caller can prompt for and inject the needed input.Pending/ResumePendingskip waiting runs — they need input, not a restart. Existing crash-resume (Resume) is unchanged, and an OTel step span for a suspend is marked OK, not errored.Why it's low-risk
Purely additive: no signature changes, no default-behavior change. A flow with no
Awaitstep behaves exactly as before.Tests
TestFlowAwaitAndResumeWith—Executesuspends cleanly at the await step (firstran once, not re-run); the run iswaiting(notpending) with correctAwait{Step,Key,Prompt};ResumeWith("approved")injects the input into the next step and the run completes withAwaitcleared.TestFlowResumeWithRejectsNonWaiting—ResumeWithon a non-waiting run errors.go build ./...,go test -race ./flow/...,go vet,golangci-lintpass.One open question (from the proposal)
Awaitergonomics here are the sentinel-return form (return flow.Await(...)as aStepFunc). If you'd rather it be a distinctAwaitStepkind in the engine, say so and I'll refactor — this PR keeps it aStepFuncso it composes withSteps(...)likeLLM/Dispatch/Call.Stages B (multi-replica leasing + idempotency) and C (durable agent tool loop) remain as described in #4816, pending your answers to the store-CAS and agent-durability-home questions there.
🤖 Generated with Claude Code
https://claude.ai/code/session_01CmdEY7pYmV5zzwCjNJ4ykL
Generated by Claude Code